nicktelford opened a new pull request, #22710:
URL: https://github.com/apache/kafka/pull/22710

   A KStream-KStream OUTER join crashes with a `SerializationException` under 
`exactly_once_v2` when `enable.transactional.statestores=true`. As soon as the 
join emits a non-joined (outer) record, the StreamThread dies and the outer 
result is never produced.
   
   `RocksDBTransactionBuffer` stages every write into a single in-memory 
`pendingWrites` map regardless of which column family the write targets. Writes 
to the offsets column family (the persisted `Position`, written under the 
`"position"` key) therefore landed in the same map as data-CF writes, so 
subsequent data-CF `range()`/`all()`/`prefixScan()` scans surfaced the offsets 
entry and tried to deserialize the serialized `Position` as a data record. The 
outer-join `ListValueStore` could not deserialize it as a value list and threw. 
The `get()` path was already gated against the offsets CF; the iterator paths 
were not. Non-transactional stores write straight to the offsets CF and never 
leak through the buffer, which is why this is transactional-only.
   
   The fix stages a write into `pendingWrites` only when its column family is 
the data column family (`cfHandle`). Offset-CF writes still go to the 
underlying `writeBatch` so they commit atomically with the data, but they no 
longer pollute data-CF iteration.
   
   A `RocksDBStoreTest` regression test asserts that `all`/`range`/`prefixScan` 
return only the data key after `writePosition()`, and a new integration test 
reproduces the original outer-join crash end to end (parameterized over 
`enable.transactional.statestores`, so the non-transactional case acts as a 
passing control).
   
   The tests in this PR originated from @bbejeck's #22709, which fixes the same 
bug via a separate `stageOffsetWrite` path; this PR takes a smaller approach 
that guards the single `stage` entry point and needs no changes to 
`RocksDBStore`.
   
   🤖 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]

Reply via email to