aliehsaeedii opened a new pull request, #22961:
URL: https://github.com/apache/kafka/pull/22961
## Summary
Adds the dual-column-family RocksDB store that lets a stream-stream
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 new store is not wired into any topology in this PR** — nothing
outside its own tests
constructs it. The change that makes the DSL select it follows in PR 2/2, so
this one can be
reviewed purely as a storage-format concern.
## Why a new store type
The outer-join store persists, per key, a `ListSerde` blob whose elements
are single serialized
values, and the element encoding differs by `dsl.store.format`:
- PLAIN: `[leftFlag(1B)][rawValue]` (a `LeftOrRightValue`)
- HEADERS: `[headersSize(varint)][headersBytes][leftFlag(1B)][rawValue]` (an
`AggregationWithHeaders<LeftOrRightValue>`)
Reading old PLAIN data through the headers serde is not merely lossy, it is
wrong in two
different ways: a right-side value is silently corrupted (the leading `0x00`
flag is read as an
empty-headers `headersSize=0` varint, dropping the first value byte, `right`
→ `ight`), and a
left-side value throws `SerializationException: negative length -1` (the
`0x01` flag is read as
`headersSize=1`).
This store cannot reuse `RocksDBTimestampedStoreWithHeaders`: that store's
whole-value
`[0x00][ts=-1]` converter would corrupt the list encoding, because here the
headers live per
element *inside* the blob rather than in front of it.
## Approach
Mirrors the KIP-1271 dual-column-family upgrade pattern, adapted to the list
store:
- `RocksDBListValueStoreWithHeaders` 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. A store with an empty DEFAULT CF opens directly on the new
column family
through a `SingleColumnFamilyAccessor`.
- `ListValueStoreUpgradeUtils.convertPlainListBlobToHeadersListBlob` is the
blob-level
converter: deserialize the list, run each element through
`HeadersBytesStore.convertToHeaderFormat`, re-serialize. Tombstones and
null list members
pass through.
- `RocksDBStore` reports a clear error on an unsupported HEADERS→PLAIN
downgrade of the new
store, mirroring the existing timestamped/headers downgrade guards.
## Testing
- `ListValueStoreUpgradeUtilsTest` — the blob converter, including the right
value that was
silently truncated and the left value that threw, plus multi-element mixed
lists, empty
lists, nulls, and the exact `0x00`-prefix byte layout.
- `RocksDBListValueStoreWithHeadersTest` — end to end against real RocksDB:
write with the
pre-headers PLAIN store, reopen the same directory as the dual-CF HEADERS
store, assert
correct reads for left/right/multi-element values, and append a new
headers-format element to
a key still living in the legacy DEFAULT column family.
- Regression: `RocksDBStoreTest`, `RocksDBTimestampedStoreWithHeadersTest`,
`ListValueStoreTest`.
Checkstyle and spotbugs pass.
`ListValueStore` is internal, so no KIP is required, and no public API
changes.
## Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
Reviewers: Matthias J. Sax <[email protected]>
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
