KAFKA-20413: Make the stream-stream outer join use the headers-aware ListValueStore [2/3] - #22966
Open
aliehsaeedii wants to merge 6 commits into
Open
KAFKA-20413: Make the stream-stream outer join use the headers-aware ListValueStore [2/3]#22966aliehsaeedii wants to merge 6 commits into
aliehsaeedii wants to merge 6 commits into
Conversation
Adds the dual-column-family RocksDB store that lets an outer-join ListValueStore written in the pre-headers PLAIN format be reopened and read in the HEADERS format (dsl.store.format=headers) without corrupting old data. The store is not wired into any topology yet; the DSL change that enables it follows in PR 2/2. The outer-join store persists, per key, a ListSerde blob whose elements are single serialized values, so it cannot reuse RocksDBTimestampedStoreWithHeaders: that store's whole-value [0x00][ts=-1] converter would corrupt the list encoding. RocksDBListValueStoreWithHeaders instead keeps legacy PLAIN blobs in the DEFAULT column family and lifts each list element to the empty-headers format on read/write (prepend 0x00 per element) via DualColumnFamilyAccessor, migrating them into a new listValueWithHeaders column family. New stores open directly on that column family through a SingleColumnFamilyAccessor. RocksDBStore reports a clear error on an unsupported HEADERS-to-PLAIN downgrade of the new store, mirroring the existing timestamped/headers downgrade guards. Tests cover the blob converter (including the right value that the naive read-through silently truncated and the left value that threw SerializationException) and an end-to-end RocksDB upgrade: write with the pre-headers PLAIN store, reopen as the dual-CF HEADERS store, and assert correct reads for left/right/multi-element values plus appends across the upgrade boundary. ListValueStore is internal, so no KIP is required. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
… format The local headers-aware list store keeps per-element headers inline, as [headersSize][headers][flag][value]. That format must not reach the changelog: the changelog topic is the only durable copy of the state, so its value format is a permanent compatibility contract. Logging the local bytes verbatim would let an old PLAIN reader -- after a version downgrade, or simply after flipping dsl.store.format back to PLAIN -- consume each element's leading empty-headers 0x00 as the LeftOrRightValue flag, silently reading left values as right ones. Do what the other KIP-1271 changelog stores do and keep the headers out of the value. Because one changelog record holds a whole list, N sets of headers have to share one record-header field, so the stripped [headersSize][headersBytes] prefixes are concatenated into a single self-delimiting blob under a reserved header rather than unpacked into individual RecordHeaders. Each chunk carries its own length, so no element count is needed and the encoding does not depend on header ordering. - ListValueStoreUpgradeUtils: add LIST_VALUE_HEADERS_HEADER_KEY, SplitListBlob, splitHeadersListBlob, joinPlainListBlobWithElementHeaders and elementHeaders. Drop the format marker: a record without the control header is a legacy record, which is exactly the all-empty-prefixes case, so the two restore paths collapse into one. - ChangeLoggingListValueBytesStoreWithHeaders: new; overrides put to log the PLAIN blob and attach the prefix header. Copies the record headers so neither the control header nor the vector clock that ProcessorContextImpl#logChange appends can leak onto the record forwarded downstream. - HeadersAwareListValueStore: new marker so StateManagerUtil picks the list-aware converter. - RecordConverters.rawListValueToHeadersListValue: re-inline the per-element headers on restore. - ListValueStoreBuilder: 5-arg ctor selecting the headers-aware changelogger. Nothing constructs any of this yet -- the DSL wiring follows separately -- so merging this changes no behaviour. Tests for ListValueStoreUpgradeUtils itself follow in a separate tests-only PR. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
…list store Wires up the storage and changelog layers added in the previous PR, and carries per-record headers through the non-joined (spurious-result) emit. - OuterStreamJoinStoreFactory: in HEADERS mode, wrap the value serde in AggregationWithHeadersSerde and build the store through the headers-aware ListValueStoreBuilder, picking the dual-column-family supplier for a persistent RocksDB store so existing PLAIN data upgrades in place. In-memory and user-supplied stores keep their supplier and are upgraded on restore by the list-aware RecordConverter. The DslStoreSuppliers request stays PLAIN because the value-with-headers stores put one headers section in front of the whole value, which is not the shape of a list. - KStreamKStreamJoin / OuterJoinStoreWrapper: read the per-element headers back out of the store and emit them on the record produced for a non-joined record, so headers survive the outer-join delay rather than being dropped. This is the only PR of the three that changes behaviour: it is the switch that makes the new store reachable. It cannot compile without PR 1. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
[1/3] dropped ListValueStoreBuilder's explicit headersFormat flag and now infers it from the supplier being a HeadersBytesStoreSupplier, so the 5-arg constructor this factory called is gone. Derive the element serde from the same signal, so HEADERS mode only takes effect when the supplier can actually provide a headers-aware bytes store. Pairing headers-format elements with the PLAIN changelogger would put each element's empty-headers 0x00 prefix on the changelog, where an old PLAIN reader takes it for the LeftOrRightValue flag. For the same reason OuterJoinStoreWrapper now reads the variant off the bytes store's HeadersAwareListValueStore marker instead of the configured DslStoreFormat, which would have claimed headers for an in-memory or user-supplied store that is in fact holding PLAIN elements. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
aliehsaeedii
force-pushed
the
headers-lvs-wiring
branch
from
July 28, 2026 18:08
add33cf to
7e36db7
Compare
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Wires the stream-stream outer join up to the headers-aware
ListValueStore, and carriesper-record headers through the non-joined (spurious-result) emit.
This is the only PR of the three that changes behaviour. It is the switch that makes the
storage and changelog layers from PR 1 reachable; it cannot compile without them.
Reviewers: Matthias J. Sax matthias@confluent.io