KAFKA-20768: Add transactional and IQ isolation to RocksDB window and session stores#22754
Open
nicktelford wants to merge 1 commit into
Open
KAFKA-20768: Add transactional and IQ isolation to RocksDB window and session stores#22754nicktelford wants to merge 1 commit into
nicktelford wants to merge 1 commit into
Conversation
… session stores The persistent RocksDB window and session stores were the last KIP-892 state stores without transactional support or isolation-level interactive queries; a READ_COMMITTED interactive query against them returned uncommitted data. Because each segment is a RocksDBStore, the write/commit path already stages and flushes per segment. This change fills the two remaining gaps in AbstractRocksDBSegmentedBytesStore: it stages the Position of uncommitted writes, merging it into the committed position atomically with the segment-buffer flush on commit, and adds an isolation-aware read view. Reads follow the same convention as RocksDBStore and the in-memory stores: the owner (stream thread) reads live via the vanilla methods, while interactive queries read through readOnly(IsolationLevel). The transaction buffer's own owner-thread check handles the non-owner READ_UNCOMMITTED snapshot, and READ_COMMITTED bypasses the buffer to read committed data. RocksDBWindowStore and RocksDBSessionStore now wrap the concrete AbstractRocksDBSegmentedBytesStore so they can reach its read view without casting. The end-to-end interactive-query isolation integration tests live on a separate branch; the persistent window and session cases there pass with this change. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Contributor
Author
kkgounder95-code
approved these changes
Jul 4, 2026
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
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.
The persistent RocksDB window and session stores were the last KIP-892
state stores without transactional support or isolation-level
interactive queries: a
READ_COMMITTEDIQ against them returned writesthat were still staged in an uncommitted transaction.
Because each segment is itself a
RocksDBStore, the write and commitpath already stages writes and flushes them per segment. This change
fills the two remaining gaps in
AbstractRocksDBSegmentedBytesStore:putnow stages thePositionof anuncommitted write in a
pendingPosition, which is merged into thecommitted position atomically with the per-segment buffer flush on
commit.getPosition()reflects staged writes (for the owner's ownview and READ_UNCOMMITTED queries), while the new
getCommittedPosition()excludes them to bound READ_COMMITTED queries.readOnly(IsolationLevel)view over thesegmented store, exposed to
RocksDBWindowStore/RocksDBSessionStoreso their
readOnly(...)views (andStoreQueryUtilsdispatch) hideuncommitted writes under READ_COMMITTED.
Reads follow the same convention as
RocksDBStoreand the in-memorystores: the owner (stream thread) reads live through the vanilla store
methods, while interactive queries read through
readOnly(IsolationLevel). The transaction buffer's own owner-threadcheck keeps the owner's reads lock-free and snapshots non-owner
READ_UNCOMMITTED reads under the read lock; READ_COMMITTED bypasses the
buffer to read committed data directly. Consequently no new isolation
surface is added to the
SegmentedBytesStoreinterface — the isolationmethods are private utilities shared between the vanilla methods and the
read view — and
RocksDBWindowStore/RocksDBSessionStorewrap theconcrete
AbstractRocksDBSegmentedBytesStoreso they reach that viewwithout casting.
This covers the standard and timestamped window stores, the session
store, and their
WithHeadersvariants by inheritance. The time-ordered/ dual-schema segmented stores (used by stream-stream joins) are out of
scope and remain non-transactional; they will be addressed as a
follow-up.
Testing:
AbstractRocksDBSegmentedBytesStoreTestgains transactionalcases parameterised over both the window and session key schemas (so
they run against
RocksDBSegmentedBytesStore,RocksDBTimestampedSegmentedBytesStore, and the with-headers variant),asserting that READ_COMMITTED hides staged writes while READ_UNCOMMITTED
exposes them, that both levels converge after commit, that the committed
position excludes staged writes until commit, and that a
non-transactional store reads identically across levels. The end-to-end
IQv1/IQv2 isolation-level integration tests live on a separate branch
and pass with this change.
🤖 Generated with Claude Code
Reviewers: kkgounder95-code (github:kkgounder95-code)