KAFKA-20413: Add missing headers-aware ListValueStore #22859
Open
aliehsaeedii wants to merge 11 commits into
Open
KAFKA-20413: Add missing headers-aware ListValueStore #22859aliehsaeedii wants to merge 11 commits into
aliehsaeedii wants to merge 11 commits into
Conversation
…ormat When dsl.store.format=headers, the stream-stream outer-join spurious-results store now reads its values as AggregationWithHeaders<LeftOrRightValue> instead of the plain LeftOrRightValue. Because that store is persistent (RocksDB) and changelogged, flipping the config on an existing application left old PLAIN records that the new headers serde mis-parsed: right-side values were silently truncated (leading 0x00 flag read as an empty-headers varint) and left-side values threw SerializationException. Migrate old data safely instead, mirroring the KIP-1271 dual-column-family pattern used by RocksDBTimestampedStoreWithHeaders, adapted to the list store whose values are ListSerde blobs of per-element encodings: - RocksDBListValueStoreWithHeaders keeps legacy PLAIN blobs in the DEFAULT column family and lifts each element to the empty-headers format on read/write (prepend 0x00 per list element) via DualColumnFamilyAccessor, migrating them into a new listValueWithHeaders column family. Selected in HEADERS mode by OuterStreamJoinStoreFactory for persistent RocksDB stores. - The changelog path (in-memory stores and RocksDB rebuilds have no local CF to migrate) is covered by a version marker: ChangeLoggingListValueBytesStoreWithHeaders stamps a reserved header on every HEADERS-format record, and a new rawListValueToHeadersListValue RecordConverter converts only unmarked (legacy) records on restore. converterForStore selects it via the HeadersAwareListValueStore marker interface (deliberately not HeadersBytesStore, which would pick the single-value headers converter and corrupt the list blob). - RocksDBStore reports a friendly error on an unsupported HEADERS-to-PLAIN downgrade of the new store. Adds unit tests for the blob converter and marker, the RecordConverter branching, converterForStore selection, and end-to-end RocksDB dual-CF upgrade tests (including append-across-the-upgrade-boundary). Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
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
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
Follow-up to #22156 (headers-aware outer-join list-value store). That PR makes the stream-stream outer-join spurious-results store read its values as
AggregationWithHeaders<LeftOrRightValue>whendsl.store.format=headers, instead of the plainLeftOrRightValue. Because that store is persistent (RocksDB) and changelogged, flipping the config on an existing application leaves old PLAIN records that the new headers serde mis-parses:0x00LeftOrRightValueflag is read as an empty-headersheadersSize=0varint, dropping the first value byte (right→ight);SerializationException: negative length -1(the0x01flag is read asheadersSize=1).This PR migrates old data safely instead of corrupting it.
Approach
Mirrors the KIP-1271 dual-column-family upgrade pattern (
RocksDBTimestampedStoreWithHeaders/DualColumnFamilyAccessor), adapted to the list store. The join processor,OuterJoinStoreWrapper, andListValueStoreare unchanged; the work is in the store factory/builder wiring plus a new bytes store.Note this store can't reuse
RocksDBTimestampedStoreWithHeadersdirectly: its values areListSerdeblobs with headers stored per element inline, so the whole-value[0x00][ts=-1]migration converter would corrupt the list encoding. Instead:RocksDBListValueStoreWithHeaders— a non-timestamped, headers-aware, dual-CF store. Legacy PLAIN blobs stay in the DEFAULT column family and are lifted to the headers format on read/write (prepend0x00per list element) viaDualColumnFamilyAccessor, migrating into a newlistValueWithHeaderscolumn family. Selected in HEADERS mode byOuterStreamJoinStoreFactoryfor persistent stores.ChangeLoggingListValueBytesStoreWithHeadersstamps a reserved marker header on every HEADERS-format record, and a newrawListValueToHeadersListValueRecordConverterconverts only unmarked (legacy) records on restore.converterForStoreselects it via theHeadersAwareListValueStoremarker interface — deliberately notHeadersBytesStore, which would pick the single-value converter and corrupt the list blob.RocksDBStorereports a clear error on an unsupported HEADERS→PLAIN downgrade of the new store.Upgrade scenarios covered
RecordConverterRecordConverterTesting
ListValueStoreUpgradeUtilsTest(blob converter + marker, incl. the previously-corrupted right value and previously-crashing left value), additions toRecordConvertersTest(marker branching / tombstone passthrough) andStateManagerUtilConverterTest(converter selection).RocksDBListValueStoreWithHeadersTest: write with the pre-headers PLAIN store, reopen as the HEADERS dual-CF store, assert correct reads for left/right/multi-element values and append-across-the-upgrade-boundary.KStreamKStreamOuterJoinTest,KStreamKStreamLeftJoinTest,RocksDBStoreTest,RocksDBTimestampedStoreWithHeadersTest,ListValueStoreTest. Checkstyle and spotbugs pass.No public API change;
ListValueStoreis internal, so no KIP is required.Committer Checklist (excluded from commit message)