nicktelford opened a new pull request, #22829: URL: https://github.com/apache/kafka/pull/22829
The time-ordered / dual-schema RocksDB stores — used for stream-stream joins and suppression buffers — were the remaining KIP-892 segmented stores without transactional support or isolation-level interactive queries. A `READ_COMMITTED` IQ against a time-ordered window or session store returned writes that were still staged in an uncommitted transaction. This was called out as an explicit follow-up in #22754, which covered the standard window and session stores. This change applies the same design to the `AbstractDualSchemaRocksDBSegmentedBytesStore` subtree (the time-ordered window and session stores and their with-headers variants). Because each segment is itself a `RocksDBStore`, the per-segment write and commit path already stages writes; the two gaps filled here are the same as before: - **Staged position** — `put` now stages the `Position` of an uncommitted write in a `pendingPosition`, merged into the committed position atomically with the per-segment buffer flush on `commit`. `getPosition()` reflects staged writes (for the owner's own view and READ_UNCOMMITTED queries), while the new `getCommittedPosition()` excludes them to bound READ_COMMITTED queries. - **Isolation-aware reads** — a `readOnly(IsolationLevel)` view over the segmented store, exposed to `RocksDBTimeOrderedWindowStore`/`RocksDBTimeOrderedSessionStore` so their `readOnly(...)` views (and `StoreQueryUtils` dispatch) hide uncommitted writes under READ_COMMITTED. Reads follow the same convention as the other stores: the owner (stream thread) reads live through the vanilla store methods, while interactive queries read through `readOnly(IsolationLevel)`. The transaction buffer's own owner-thread check keeps the owner's reads lock-free and snapshots non-owner READ_UNCOMMITTED reads under the read lock; READ_COMMITTED bypasses the buffer. As before, no isolation surface is added to the `SegmentedBytesStore` interface — the isolation methods are package-private utilities shared between the vanilla methods and the read view — and the wrapping stores hold the concrete `AbstractRocksDBTimeOrderedSegmentedBytesStore` so they reach that view without casting. The one wrinkle specific to the dual-schema stores is the index-to-base lookup: an indexed store iterates the key-ordered index and then reads each value from the time-ordered base store. That base read now goes through the same isolation view as the index scan, so both sides observe a consistent snapshot. The base read on the owner path also repairs stale index entries; because an interactive query must never mutate the store, that repair is skipped when reading through an isolation view. The plain time-ordered key-value store (the suppression buffer backing store) is not queried interactively, but it inherits the staged position and commit changes for free and remains correct. Testing: `AbstractDualSchemaRocksDBSegmentedBytesStoreTest` gains transactional cases that run against all four concrete configurations (window and session, each with and without an index), 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. 🤖 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]
