KAFKA-20749: Stop offset writes leaking into data iteration#22710
Merged
bbejeck merged 2 commits intoJun 30, 2026
Conversation
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) therefore landed in the same map as data-CF writes, so subsequent data-CF scans surfaced the offsets "position" entry and tried to deserialize it as a data record, throwing a SerializationException. Stage a write into pendingWrites only when its column family is the data column family (cfHandle); offset-CF writes still go to the underlying writeBatch but no longer pollute data-CF iteration. A RocksDBStoreTest regression test asserts that all/range/prefixScan return only the data key after writePosition(), and an integration test reproduces the original outer-join crash end to end. Co-Authored-By: Bill Bejeck <bbejeck@apache.org> Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Member
|
Thanks @nicktelford the |
stage(cf, key, value) no longer mirrors non-primary-CF writes into pendingWrites, so buffer.get() correctly returns null for keys that were only staged to a secondary column family. Co-Authored-By: Claude Sonnet 4.6 (1M context) <noreply@anthropic.com>
Contributor
Author
@bbejeck Good catch. Fixed! |
Member
|
Merged #22710 into trunk |
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
2 participants
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.
A KStream-KStream OUTER join crashes with a
SerializationExceptionunder
exactly_once_v2whenenable.transactional.statestores=true. Assoon as the join emits a non-joined (outer) record, the StreamThread
dies and the outer result is never produced.
RocksDBTransactionBufferstages every write into a single in-memorypendingWritesmap regardless of which column family the write targets.Writes to the offsets column family (the persisted
Position, writtenunder the
"position"key) therefore landed in the same map as data-CFwrites, so subsequent data-CF
range()/all()/prefixScan()scanssurfaced the offsets entry and tried to deserialize the serialized
Positionas a data record. The outer-joinListValueStorecould notdeserialize it as a value list and threw. The
get()path was alreadygated 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
pendingWritesonly when its column familyis the data column family (
cfHandle). Offset-CF writes still go to theunderlying
writeBatchso they commit atomically with the data, butthey no longer pollute data-CF iteration.
A
RocksDBStoreTestregression test asserts thatall/range/prefixScanreturn only the data key afterwritePosition(), and a new integration test reproduces the originalouter-join crash end to end (parameterized over
enable.transactional.statestores, so the non-transactional case actsas a passing control).
The tests in this PR originated from @bbejeck's #22709, which fixes the
same bug via a separate
stageOffsetWritepath; this PR takes a smallerapproach that guards the single
stageentry point and needs no changesto
RocksDBStore.🤖 Generated with Claude Code
Reviewers: Bill Bejeck bbejeck@apache.org