aliehsaeedii opened a new pull request, #22859: URL: https://github.com/apache/kafka/pull/22859
## 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>` when `dsl.store.format=headers`, instead of the plain `LeftOrRightValue`. 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: - **right-side** values are silently corrupted — the leading `0x00` `LeftOrRightValue` flag is read as an empty-headers `headersSize=0` varint, dropping the first value byte (`right` → `ight`); - **left-side** values throw `SerializationException: negative length -1` (the `0x01` flag is read as `headersSize=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`, and `ListValueStore` are unchanged; the work is in the store factory/builder wiring plus a new bytes store. Note this store can't reuse `RocksDBTimestampedStoreWithHeaders` directly: its values are `ListSerde` blobs 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 (prepend `0x00` per **list element**) via `DualColumnFamilyAccessor`, migrating into a new `listValueWithHeaders` column family. Selected in HEADERS mode by `OuterStreamJoinStoreFactory` for persistent stores. - **Changelog restore path** (in-memory stores and RocksDB rebuilds after state loss have no local CF to migrate): `ChangeLoggingListValueBytesStoreWithHeaders` stamps a reserved marker 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 converter and corrupt the list blob. - **`RocksDBStore`** reports a clear error on an unsupported HEADERS→PLAIN downgrade of the new store. ## Upgrade scenarios covered | Scenario | Mechanism | |---|---| | Rolling upgrade, RocksDB, retained local state | dual-CF migrate-on-read/write | | RocksDB, lost/wiped state → full restore | changelog marker + `RecordConverter` | | In-memory store (no CF) → restore | changelog marker + `RecordConverter` | | Fresh HEADERS store | records marked, converter is a no-op | | Staying PLAIN | unchanged | ## Testing - New unit tests: `ListValueStoreUpgradeUtilsTest` (blob converter + marker, incl. the previously-corrupted right value and previously-crashing left value), additions to `RecordConvertersTest` (marker branching / tombstone passthrough) and `StateManagerUtilConverterTest` (converter selection). - New end-to-end store test `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. - Regression: `KStreamKStreamOuterJoinTest`, `KStreamKStreamLeftJoinTest`, `RocksDBStoreTest`, `RocksDBTimestampedStoreWithHeadersTest`, `ListValueStoreTest`. Checkstyle and spotbugs pass. No public API change; `ListValueStore` is internal, so no KIP is required. ## Committer Checklist (excluded from commit message) - [ ] Verify design and implementation - [ ] Verify test coverage and CI build status - [ ] Verify documentation (including upgrade notes) -- 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]
