KAFKA-20413: Add list-value store format and headers round-trip tests [3/3] - #22967
Open
aliehsaeedii wants to merge 3 commits into
Open
KAFKA-20413: Add list-value store format and headers round-trip tests [3/3]#22967aliehsaeedii wants to merge 3 commits into
aliehsaeedii wants to merge 3 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>
Test-only. Covers the pieces the first two PRs left tested transitively, and fills a pre-existing gap in list-store coverage. - ListValueStoreUpgradeUtilsTest: direct coverage of the changelog encoding -- the PLAIN-to-HEADERS blob conversion, the split/join pair, the exact byte layout (the zigzag headersSize varint, one 0x00 per element with no headers), duplicate header keys across elements, tombstones, empty lists, independence from record-header ordering, and the error paths for truncated, trailing and oversized header prefixes. - ListValueStoreTest: parametrize the existing list/iterator invariants over both plain and headers-embedding value serdes. - ListValueStoreWithHeadersTest: assert that per-record headers and timestamps survive a round-trip through the builder -> metered -> change-logger -> ListValueStore pipeline. Co-Authored-By: Claude Opus 5 (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
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
Test-only. Adds direct coverage for the changelog encoding introduced in #22961, and fills a
pre-existing gap in
ListValueStorecoverage. No production code changes.Reviewers: Matthias J. Sax matthias@confluent.io