This is an automated email from the ASF dual-hosted git repository.
bbejeck pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 704fcb26554 KAFKA-20768: Add transactional and IQ isolation to RocksDB
window and session stores (#22754)
704fcb26554 is described below
commit 704fcb265543cad94e04eac9d637133735e2298a
Author: Nick Telford <[email protected]>
AuthorDate: Tue Jul 14 14:02:19 2026 +0100
KAFKA-20768: Add transactional and IQ isolation to RocksDB window and
session stores (#22754)
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` IQ against them returned writes
that were still staged in an uncommitted transaction.
Because each segment is itself a `RocksDBStore`, the write and commit
path already stages writes and flushes them per segment. This change
fills the two remaining gaps in `AbstractRocksDBSegmentedBytesStore`:
- **Staged position** — `put` now stages the `Position` of an
uncommitted write in a `pendingPosition`, which is 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 `RocksDBWindowStore`/`RocksDBSessionStore`
so their `readOnly(...)` views (and `StoreQueryUtils` dispatch) hide
uncommitted writes under READ_COMMITTED.
Reads follow the same convention as `RocksDBStore` and the in-memory
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 to read committed data directly. Consequently no new isolation
surface is added to the `SegmentedBytesStore` interface — the isolation
methods are private utilities shared between the vanilla methods and the
read view — and `RocksDBWindowStore`/`RocksDBSessionStore` wrap the
concrete `AbstractRocksDBSegmentedBytesStore` so they reach that view
without casting.
This covers the standard and timestamped window stores, the session
store, and their `WithHeaders` variants 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: `AbstractRocksDBSegmentedBytesStoreTest` gains transactional
cases 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.
Reviewers: kkgounder95-code (github:kkgounder95-code), Bill Bejeck
<[email protected]>
---
.../AbstractRocksDBSegmentedBytesStore.java | 188 +++++++++++++++------
.../state/internals/RocksDBSessionStore.java | 132 +++++++++++----
.../internals/RocksDBSessionStoreWithHeaders.java | 2 +-
.../streams/state/internals/RocksDBStore.java | 1 +
.../internals/RocksDBTimestampedWindowStore.java | 2 +-
.../RocksDBTimestampedWindowStoreWithHeaders.java | 2 +-
.../state/internals/RocksDBWindowStore.java | 109 ++++++++++--
.../streams/state/internals/SegmentIterator.java | 23 ++-
.../AbstractRocksDBSegmentedBytesStoreTest.java | 118 +++++++++++++
.../PlainToHeadersWindowStoreAdapterTest.java | 2 +-
.../RocksDBSessionStoreWithHeadersTest.java | 2 +-
...cksDBTimestampedWindowStoreWithHeadersTest.java | 2 +-
...TimestampedToHeadersWindowStoreAdapterTest.java | 2 +-
13 files changed, 473 insertions(+), 112 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
index 9357583ec53..0d3c25783ae 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStore.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.metrics.Sensor;
import org.apache.kafka.common.utils.Bytes;
@@ -44,6 +45,7 @@ import java.util.Collection;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
import static
org.apache.kafka.streams.StreamsConfig.InternalConfig.IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED;
import static
org.apache.kafka.streams.processor.internals.ProcessorContextUtils.asInternalProcessorContext;
@@ -61,6 +63,9 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
private long observedStreamTime = ConsumerRecord.NO_TIMESTAMP;
private boolean consistencyEnabled = false;
private Position position;
+ // Position of writes staged in the current transaction; merged into the
committed position on commit().
+ private Position pendingPosition = Position.emptyPosition();
+ private boolean transactional;
private volatile boolean open;
AbstractRocksDBSegmentedBytesStore(final String name,
@@ -82,20 +87,21 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
public KeyValueIterator<Bytes, byte[]> fetch(final Bytes key,
final long from,
final long to) {
- return fetch(key, from, to, true);
+ return fetch(key, from, to, true, null);
}
@Override
public KeyValueIterator<Bytes, byte[]> backwardFetch(final Bytes key,
final long from,
final long to) {
- return fetch(key, from, to, false);
+ return fetch(key, from, to, false, null);
}
- KeyValueIterator<Bytes, byte[]> fetch(final Bytes key,
- final long from,
- final long to,
- final boolean forward) {
+ private KeyValueIterator<Bytes, byte[]> fetch(final Bytes key,
+ final long from,
+ final long to,
+ final boolean forward,
+ final IsolationLevel
isolationLevel) {
final long actualFrom = getActualFrom(from);
if (keySchema instanceof WindowKeySchema && to < actualFrom) {
@@ -113,7 +119,8 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
keySchema.hasNextCondition(key, key, actualFrom, to, forward),
binaryFrom,
binaryTo,
- forward);
+ forward,
+ isolationLevel);
}
private long getActualFrom(final long from) {
@@ -125,7 +132,7 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
final Bytes keyTo,
final long from,
final long to) {
- return fetch(keyFrom, keyTo, from, to, true);
+ return fetch(keyFrom, keyTo, from, to, true, null);
}
@Override
@@ -133,14 +140,15 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
final Bytes keyTo,
final long from,
final long to) {
- return fetch(keyFrom, keyTo, from, to, false);
+ return fetch(keyFrom, keyTo, from, to, false, null);
}
- KeyValueIterator<Bytes, byte[]> fetch(final Bytes keyFrom,
- final Bytes keyTo,
- final long from,
- final long to,
- final boolean forward) {
+ private KeyValueIterator<Bytes, byte[]> fetch(final Bytes keyFrom,
+ final Bytes keyTo,
+ final long from,
+ final long to,
+ final boolean forward,
+ final IsolationLevel
isolationLevel) {
if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
LOG.warn("Returning empty iterator for fetch with invalid key
range: from > to. " +
"This may be due to range arguments set in the wrong
order, " +
@@ -166,59 +174,49 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
keySchema.hasNextCondition(keyFrom, keyTo, actualFrom, to,
forward),
binaryFrom,
binaryTo,
- forward);
+ forward,
+ isolationLevel);
}
@Override
public KeyValueIterator<Bytes, byte[]> all() {
- final long actualFrom = getActualFrom(0);
- final List<S> searchSpace = keySchema.segmentsToSearch(segments,
actualFrom, Long.MAX_VALUE, true);
-
- return new SegmentIterator<>(
- searchSpace.iterator(),
- keySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, true),
- null,
- null,
- true);
+ return all(true, null);
}
@Override
public KeyValueIterator<Bytes, byte[]> backwardAll() {
- final long actualFrom = getActualFrom(0);
+ return all(false, null);
+ }
- final List<S> searchSpace = keySchema.segmentsToSearch(segments,
actualFrom, Long.MAX_VALUE, false);
+ private KeyValueIterator<Bytes, byte[]> all(final boolean forward, final
IsolationLevel isolationLevel) {
+ final long actualFrom = getActualFrom(0);
+ final List<S> searchSpace = keySchema.segmentsToSearch(segments,
actualFrom, Long.MAX_VALUE, forward);
return new SegmentIterator<>(
searchSpace.iterator(),
- keySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, false),
+ keySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, forward),
null,
null,
- false);
+ forward,
+ isolationLevel);
}
@Override
public KeyValueIterator<Bytes, byte[]> fetchAll(final long timeFrom,
final long timeTo) {
- final long actualFrom = getActualFrom(timeFrom);
-
- if (keySchema instanceof WindowKeySchema && timeTo < actualFrom) {
- LOG.debug("Returning no records for as timeTo ({}) < actualFrom
({}) ", timeTo, actualFrom);
- return KeyValueIterators.emptyIterator();
- }
-
- final List<S> searchSpace = segments.segments(actualFrom, timeTo,
true);
-
- return new SegmentIterator<>(
- searchSpace.iterator(),
- keySchema.hasNextCondition(null, null, actualFrom, timeTo,
true),
- null,
- null,
- true);
+ return fetchAll(timeFrom, timeTo, true, null);
}
@Override
public KeyValueIterator<Bytes, byte[]> backwardFetchAll(final long
timeFrom,
final long timeTo)
{
+ return fetchAll(timeFrom, timeTo, false, null);
+ }
+
+ private KeyValueIterator<Bytes, byte[]> fetchAll(final long timeFrom,
+ final long timeTo,
+ final boolean forward,
+ final IsolationLevel
isolationLevel) {
final long actualFrom = getActualFrom(timeFrom);
if (keySchema instanceof WindowKeySchema && timeTo < actualFrom) {
@@ -226,14 +224,15 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
return KeyValueIterators.emptyIterator();
}
- final List<S> searchSpace = segments.segments(actualFrom, timeTo,
false);
+ final List<S> searchSpace = segments.segments(actualFrom, timeTo,
forward);
return new SegmentIterator<>(
searchSpace.iterator(),
- keySchema.hasNextCondition(null, null, actualFrom, timeTo,
false),
+ keySchema.hasNextCondition(null, null, actualFrom, timeTo,
forward),
null,
null,
- false);
+ forward,
+ isolationLevel);
}
@Override
@@ -258,7 +257,8 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
expiredRecordSensor.record(1.0d,
internalProcessorContext.currentSystemTimeMs());
} else {
synchronized (position) {
- StoreQueryUtils.updatePosition(position,
internalProcessorContext);
+ // Transactional puts stage their position too, so
READ_COMMITTED never sees a position ahead of the data.
+ StoreQueryUtils.updatePosition(transactional ? pendingPosition
: position, internalProcessorContext);
segment.put(key, value);
}
}
@@ -266,6 +266,10 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
@Override
public byte[] get(final Bytes key) {
+ return get(key, null);
+ }
+
+ private byte[] get(final Bytes key, final IsolationLevel isolationLevel) {
final long timestampFromKey = keySchema.segmentTimestamp(key);
// check if timestamp is expired
if (timestampFromKey < observedStreamTime - retentionPeriod + 1) {
@@ -277,7 +281,14 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
if (segment == null) {
return null;
}
- return segment.get(key);
+ // Live read for the vanilla path; interactive queries read through
the segment's isolation view.
+ return isolationLevel == null ? segment.get(key) :
segment.readOnly(isolationLevel).get(key);
+ }
+
+ /** An isolation-bound read view; the level is resolved once here and
reads share the store's private utilities. */
+ ReadOnlyView readOnly(final IsolationLevel isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(this, isolationLevel);
}
@Override
@@ -301,6 +312,7 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
segments.openExisting(internalProcessorContext, observedStreamTime);
this.position = segments.position;
+ this.pendingPosition = Position.emptyPosition();
StoreQueryUtils.maybeMigrateExistingPositionFile(stateStoreContext.stateDir(),
name(), this.position);
// register and possibly restore the state from the logs
@@ -316,11 +328,23 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
stateStoreContext.appConfigs(),
IQ_CONSISTENCY_OFFSET_VECTOR_ENABLED,
false);
+
+ transactional = StreamsConfig.InternalConfig.getBoolean(
+ stateStoreContext.appConfigs(),
+ StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
+ false);
}
@Override
public void commit(final Map<TopicPartition, Long> changelogOffsets) {
- segments.commit(changelogOffsets);
+ synchronized (position) {
+ if (transactional) {
+ // Publish the staged position atomically with the
segment-buffer flush below.
+ position.merge(pendingPosition);
+ pendingPosition = Position.emptyPosition();
+ }
+ segments.commit(changelogOffsets);
+ }
}
@SuppressWarnings("deprecation")
@@ -347,6 +371,8 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
public void close() {
open = false;
segments.close();
+ // The segments discard their uncommitted writes on close; drop the
staged position to match.
+ pendingPosition = Position.emptyPosition();
}
@Override
@@ -415,6 +441,68 @@ public class AbstractRocksDBSegmentedBytesStore<S extends
Segment> implements Se
@Override
public Position getPosition() {
+ // Include staged writes so the owner's view (and READ_UNCOMMITTED
queries) stays consistent.
+ if (transactional) {
+ synchronized (position) {
+ return position.copy().merge(pendingPosition);
+ }
+ }
return position;
}
-}
\ No newline at end of file
+
+ Position getCommittedPosition() {
+ if (transactional) {
+ synchronized (position) {
+ return position.copy();
+ }
+ }
+ return position;
+ }
+
+ /** Read view bound to an {@link IsolationLevel}, delegating to the
store's private read utilities. */
+ static final class ReadOnlyView {
+ private final AbstractRocksDBSegmentedBytesStore<?> store;
+ private final IsolationLevel isolationLevel;
+
+ private ReadOnlyView(final AbstractRocksDBSegmentedBytesStore<?>
store, final IsolationLevel isolationLevel) {
+ this.store = store;
+ this.isolationLevel = isolationLevel;
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetch(final Bytes key, final long
from, final long to) {
+ return store.fetch(key, from, to, true, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> backwardFetch(final Bytes key, final
long from, final long to) {
+ return store.fetch(key, from, to, false, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetch(final Bytes keyFrom, final Bytes
keyTo, final long from, final long to) {
+ return store.fetch(keyFrom, keyTo, from, to, true, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> backwardFetch(final Bytes keyFrom,
final Bytes keyTo, final long from, final long to) {
+ return store.fetch(keyFrom, keyTo, from, to, false,
isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> all() {
+ return store.all(true, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> backwardAll() {
+ return store.all(false, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetchAll(final long from, final long
to) {
+ return store.fetchAll(from, to, true, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> backwardFetchAll(final long from,
final long to) {
+ return store.fetchAll(from, to, false, isolationLevel);
+ }
+
+ byte[] get(final Bytes key) {
+ return store.get(key, isolationLevel);
+ }
+ }
+}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
index 80af230c300..da52bac22be 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStore.java
@@ -16,25 +16,31 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.query.PositionBound;
import org.apache.kafka.streams.query.Query;
import org.apache.kafka.streams.query.QueryConfig;
import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlySessionStore;
import org.apache.kafka.streams.state.SessionStore;
+import java.time.Instant;
+import java.util.Objects;
+
public class RocksDBSessionStore
- extends WrappedStateStore<SegmentedBytesStore, Object, Object>
+ extends WrappedStateStore<AbstractRocksDBSegmentedBytesStore<?>, Object,
Object>
implements SessionStore<Bytes, byte[]> {
private StateStoreContext stateStoreContext;
- RocksDBSessionStore(final SegmentedBytesStore bytesStore) {
+ RocksDBSessionStore(final AbstractRocksDBSegmentedBytesStore<?>
bytesStore) {
super(bytesStore);
}
@@ -49,12 +55,15 @@ public class RocksDBSessionStore
final PositionBound positionBound,
final QueryConfig config) {
+ final Position queryPosition = config.getIsolationLevel() ==
IsolationLevel.READ_COMMITTED
+ ? wrapped().getCommittedPosition()
+ : getPosition();
return StoreQueryUtils.handleBasicQueries(
query,
positionBound,
config,
this,
- getPosition(),
+ queryPosition,
stateStoreContext
);
}
@@ -63,24 +72,14 @@ public class RocksDBSessionStore
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes
key,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator = wrapped().fetch(
- key,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator);
+ return sessionIterator(wrapped().fetch(key, earliestSessionEndTime,
latestSessionStartTime));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetch(
- key,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator);
+ return sessionIterator(wrapped().backwardFetch(key,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
@@ -88,13 +87,7 @@ public class RocksDBSessionStore
final Bytes
keyTo,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator = wrapped().fetch(
- keyFrom,
- keyTo,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator);
+ return sessionIterator(wrapped().fetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
@@ -102,24 +95,19 @@ public class RocksDBSessionStore
final Bytes keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetch(
- keyFrom,
- keyTo,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator);
+ return sessionIterator(wrapped().backwardFetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
public byte[] fetchSession(final Bytes key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
- return wrapped().get(SessionKeySchema.toBinary(
- key,
- earliestSessionEndTime,
- latestSessionStartTime
- ));
+ return wrapped().get(SessionKeySchema.toBinary(key,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ // Shared by the live methods above and the ReadOnlyView below; only the
iterator's source differs.
+ private static KeyValueIterator<Windowed<Bytes>, byte[]>
sessionIterator(final KeyValueIterator<Bytes, byte[]> bytesIterator) {
+ return new WrappedSessionStoreIterator(bytesIterator);
}
@Override
@@ -151,4 +139,80 @@ public class RocksDBSessionStore
public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
wrapped().put(SessionKeySchema.toBinary(sessionKey), aggregate);
}
+
+ @Override
+ public ReadOnlySessionStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /** Read view; reads go through the segmented store's isolation view, so
READ_COMMITTED hides staged writes. */
+ private final class ReadOnlyView implements ReadOnlySessionStore<Bytes,
byte[]> {
+
+ private final AbstractRocksDBSegmentedBytesStore.ReadOnlyView
segmented;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.segmented = wrapped().readOnly(isolationLevel);
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final long
earliestSessionEndTime, final long latestSessionStartTime) {
+ return segmented.get(SessionKeySchema.toBinary(key,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final Instant
earliestSessionEndTime, final Instant latestSessionStartTime) {
+ return fetchSession(key, earliestSessionEndTime.toEpochMilli(),
latestSessionStartTime.toEpochMilli());
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final
Bytes key,
+ final
long earliestSessionEndTime,
+ final
long latestSessionStartTime) {
+ return sessionIterator(segmented.fetch(key,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes key,
+
final long earliestSessionEndTime,
+
final long latestSessionStartTime) {
+ return sessionIterator(segmented.backwardFetch(key,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final
Bytes keyFrom,
+ final
Bytes keyTo,
+ final
long earliestSessionEndTime,
+ final
long latestSessionStartTime) {
+ return sessionIterator(segmented.fetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes keyFrom,
+
final Bytes keyTo,
+
final long earliestSessionEndTime,
+
final long latestSessionStartTime) {
+ return sessionIterator(segmented.backwardFetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
key) {
+ return findSessions(key, 0, Long.MAX_VALUE);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes key) {
+ return backwardFindSessions(key, 0, Long.MAX_VALUE);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo) {
+ return findSessions(keyFrom, keyTo, 0, Long.MAX_VALUE);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes keyFrom, final Bytes keyTo) {
+ return backwardFindSessions(keyFrom, keyTo, 0, Long.MAX_VALUE);
+ }
+ }
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeaders.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeaders.java
index 7e171ba952a..91fd584523d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeaders.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeaders.java
@@ -37,7 +37,7 @@ import org.apache.kafka.streams.state.HeadersBytesStore;
*/
class RocksDBSessionStoreWithHeaders extends RocksDBSessionStore implements
HeadersBytesStore {
- RocksDBSessionStoreWithHeaders(final SegmentedBytesStore bytesStore) {
+ RocksDBSessionStoreWithHeaders(final AbstractRocksDBSegmentedBytesStore<?>
bytesStore) {
super(bytesStore);
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index e7b2eef3a4c..398ce2f88d0 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -756,6 +756,7 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
@Override
public ReadOnlyKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ validateStoreOpen();
return new ReadOnlyView(dbAccessor.readOnly(isolationLevel));
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStore.java
index b96748eed1a..f6d3ef1017d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStore.java
@@ -20,7 +20,7 @@ import org.apache.kafka.streams.state.TimestampedBytesStore;
class RocksDBTimestampedWindowStore extends RocksDBWindowStore implements
TimestampedBytesStore {
- RocksDBTimestampedWindowStore(final SegmentedBytesStore bytesStore,
+ RocksDBTimestampedWindowStore(final AbstractRocksDBSegmentedBytesStore<?>
bytesStore,
final boolean retainDuplicates,
final long windowSize) {
super(bytesStore, retainDuplicates, windowSize);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeaders.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeaders.java
index 37b1994b02f..fa22aa50907 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeaders.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeaders.java
@@ -47,7 +47,7 @@ import org.apache.kafka.streams.state.TimestampedBytesStore;
*/
class RocksDBTimestampedWindowStoreWithHeaders extends RocksDBWindowStore
implements TimestampedBytesStore, HeadersBytesStore {
- RocksDBTimestampedWindowStoreWithHeaders(final SegmentedBytesStore
bytesStore,
+ RocksDBTimestampedWindowStoreWithHeaders(final
AbstractRocksDBSegmentedBytesStore<?> bytesStore,
final boolean retainDuplicates,
final long windowSize) {
super(bytesStore, retainDuplicates, windowSize);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
index 61823c3f261..ede47b4b76a 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBWindowStore.java
@@ -16,20 +16,26 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.processor.StateStore;
import org.apache.kafka.streams.processor.StateStoreContext;
+import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.query.PositionBound;
import org.apache.kafka.streams.query.Query;
import org.apache.kafka.streams.query.QueryConfig;
import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
+import java.time.Instant;
+import java.util.Objects;
+
public class RocksDBWindowStore
- extends WrappedStateStore<SegmentedBytesStore, Object, Object>
+ extends WrappedStateStore<AbstractRocksDBSegmentedBytesStore<?>, Object,
Object>
implements WindowStore<Bytes, byte[]> {
private final boolean retainDuplicates;
@@ -39,7 +45,7 @@ public class RocksDBWindowStore
private StateStoreContext stateStoreContext;
- RocksDBWindowStore(final SegmentedBytesStore bytesStore,
+ RocksDBWindowStore(final AbstractRocksDBSegmentedBytesStore<?> bytesStore,
final boolean retainDuplicates,
final long windowSize) {
super(bytesStore);
@@ -69,14 +75,12 @@ public class RocksDBWindowStore
@Override
public WindowStoreIterator<byte[]> fetch(final Bytes key, final long
timeFrom, final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().fetch(key, timeFrom, timeTo);
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).valuesIterator();
+ return valuesIterator(wrapped().fetch(key, timeFrom, timeTo));
}
@Override
public WindowStoreIterator<byte[]> backwardFetch(final Bytes key, final
long timeFrom, final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetch(key, timeFrom, timeTo);
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).valuesIterator();
+ return valuesIterator(wrapped().backwardFetch(key, timeFrom, timeTo));
}
@Override
@@ -84,8 +88,7 @@ public class RocksDBWindowStore
final Bytes keyTo,
final long timeFrom,
final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().fetch(keyFrom, keyTo, timeFrom, timeTo);
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
+ return keyValueIterator(wrapped().fetch(keyFrom, keyTo, timeFrom,
timeTo));
}
@Override
@@ -93,45 +96,115 @@ public class RocksDBWindowStore
final Bytes
keyTo,
final long
timeFrom,
final long
timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetch(keyFrom, keyTo, timeFrom, timeTo);
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
+ return keyValueIterator(wrapped().backwardFetch(keyFrom, keyTo,
timeFrom, timeTo));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
- final KeyValueIterator<Bytes, byte[]> bytesIterator = wrapped().all();
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
+ return keyValueIterator(wrapped().all());
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardAll();
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
+ return keyValueIterator(wrapped().backwardAll());
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long
timeFrom, final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().fetchAll(timeFrom, timeTo);
- return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
+ return keyValueIterator(wrapped().fetchAll(timeFrom, timeTo));
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final
long timeFrom, final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetchAll(timeFrom, timeTo);
+ return keyValueIterator(wrapped().backwardFetchAll(timeFrom, timeTo));
+ }
+
+ // Shared by the live methods above and the ReadOnlyView below; only the
iterator's source differs.
+ private WindowStoreIterator<byte[]> valuesIterator(final
KeyValueIterator<Bytes, byte[]> bytesIterator) {
+ return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).valuesIterator();
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> keyValueIterator(final
KeyValueIterator<Bytes, byte[]> bytesIterator) {
return new WindowStoreIteratorWrapper(bytesIterator,
windowSize).keyValueIterator();
}
+ @Override
+ public ReadOnlyWindowStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /** Read view; reads go through the segmented store's isolation view, so
READ_COMMITTED hides staged writes. */
+ private final class ReadOnlyView implements ReadOnlyWindowStore<Bytes,
byte[]> {
+
+ private final AbstractRocksDBSegmentedBytesStore.ReadOnlyView
segmented;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.segmented = wrapped().readOnly(isolationLevel);
+ }
+
+ @Override
+ public byte[] fetch(final Bytes key, final long time) {
+ return segmented.get(WindowKeySchema.toStoreKeyBinary(key, time,
seqnum));
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> fetch(final Bytes key, final
Instant timeFrom, final Instant timeTo) {
+ return valuesIterator(segmented.fetch(key,
timeFrom.toEpochMilli(), timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> backwardFetch(final Bytes key,
final Instant timeFrom, final Instant timeTo) {
+ return valuesIterator(segmented.backwardFetch(key,
timeFrom.toEpochMilli(), timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo,
+ final Instant
timeFrom, final Instant timeTo) {
+ return keyValueIterator(segmented.fetch(keyFrom, keyTo,
timeFrom.toEpochMilli(), timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes keyFrom, final Bytes keyTo,
+ final
Instant timeFrom, final Instant timeTo) {
+ return keyValueIterator(segmented.backwardFetch(keyFrom, keyTo,
timeFrom.toEpochMilli(), timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final
Instant timeFrom, final Instant timeTo) {
+ return
keyValueIterator(segmented.fetchAll(timeFrom.toEpochMilli(),
timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFetchAll(final Instant timeFrom, final Instant timeTo) {
+ return
keyValueIterator(segmented.backwardFetchAll(timeFrom.toEpochMilli(),
timeTo.toEpochMilli()));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
+ return keyValueIterator(segmented.all());
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
+ return keyValueIterator(segmented.backwardAll());
+ }
+ }
+
@Override
public <R> QueryResult<R> query(final Query<R> query,
final PositionBound positionBound,
final QueryConfig config) {
+ final Position queryPosition = config.getIsolationLevel() ==
IsolationLevel.READ_COMMITTED
+ ? wrapped().getCommittedPosition()
+ : getPosition();
return StoreQueryUtils.handleBasicQueries(
query,
positionBound,
config,
this,
- getPosition(),
+ queryPosition,
stateStoreContext
);
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
index a0c52db873b..d45e950b535 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/SegmentIterator.java
@@ -16,10 +16,12 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import java.util.Iterator;
import java.util.NoSuchElementException;
@@ -32,6 +34,9 @@ public class SegmentIterator<S extends Segment> implements
KeyValueIterator<Byte
private final Bytes from;
private final Bytes to;
private final boolean forward;
+ // Non-null only for interactive-query iteration, where reads go through
the segment's isolation view;
+ // null means read the segment live (the owner's own path during
processing).
+ private final IsolationLevel isolationLevel;
protected final Iterator<S> segments;
protected final HasNextCondition hasNextCondition;
@@ -43,11 +48,21 @@ public class SegmentIterator<S extends Segment> implements
KeyValueIterator<Byte
final Bytes from,
final Bytes to,
final boolean forward) {
+ this(segments, hasNextCondition, from, to, forward, null);
+ }
+
+ SegmentIterator(final Iterator<S> segments,
+ final HasNextCondition hasNextCondition,
+ final Bytes from,
+ final Bytes to,
+ final boolean forward,
+ final IsolationLevel isolationLevel) {
this.segments = segments;
this.hasNextCondition = hasNextCondition;
this.from = from;
this.to = to;
this.forward = forward;
+ this.isolationLevel = isolationLevel;
}
@Override
@@ -74,10 +89,12 @@ public class SegmentIterator<S extends Segment> implements
KeyValueIterator<Byte
close();
currentSegment = segments.next();
try {
- if (forward) {
- currentIterator = currentSegment.range(from, to);
+ if (isolationLevel == null) {
+ currentIterator = forward ? currentSegment.range(from, to)
: currentSegment.reverseRange(from, to);
} else {
- currentIterator = currentSegment.reverseRange(from, to);
+ // Interactive query: read through the segment's isolation
view.
+ final ReadOnlyKeyValueStore<Bytes, byte[]> segmentView =
currentSegment.readOnly(isolationLevel);
+ currentIterator = forward ? segmentView.range(from, to) :
segmentView.reverseRange(from, to);
}
} catch (final InvalidStateStoreException e) {
// segment may have been closed so we ignore it.
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStoreTest.java
index f1712f2bf90..acd6d219cc9 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractRocksDBSegmentedBytesStoreTest.java
@@ -17,6 +17,7 @@
package org.apache.kafka.streams.state.internals;
import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.Metric;
import org.apache.kafka.common.MetricName;
import org.apache.kafka.common.header.Headers;
@@ -26,6 +27,7 @@ import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.record.TimestampType;
import org.apache.kafka.common.record.internal.RecordBatch;
import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
@@ -94,6 +96,7 @@ import static org.hamcrest.Matchers.hasEntry;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -1006,6 +1009,121 @@ public abstract class
AbstractRocksDBSegmentedBytesStoreTest<S extends Segment>
return Set.of(Objects.requireNonNull(windowDir.list()));
}
+ @ParameterizedTest
+ @MethodSource("getKeySchemas")
+ public void readCommittedShouldHideStagedWritesUntilCommit(final
SegmentedBytesStore.KeySchema schema) {
+ initTransactional(schema);
+ final String key = "a";
+ final Bytes storeKey = serializeKey(new Windowed<>(key, windows[0]));
+
+ bytesStore.put(storeKey, serializeValue(10L));
+ bytesStore.commit(Map.of());
+
+ // Stage an overwrite that is not yet committed.
+ bytesStore.put(storeKey, serializeValue(50L));
+
+ // Point get: READ_UNCOMMITTED sees the staged value; READ_COMMITTED
sees the last committed value.
+ assertEquals(50L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ assertEquals(10L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_COMMITTED).get(storeKey)));
+
+ // Range fetch (exercises SegmentIterator's isolation routing) must
respect the same visibility.
+ assertEquals(
+ List.of(KeyValue.pair(new Windowed<>(key, windows[0]), 50L)),
+
toListAndCloseIterator(bytesStore.readOnly(IsolationLevel.READ_UNCOMMITTED).fetch(Bytes.wrap(key.getBytes()),
0, windows[0].start())));
+ assertEquals(
+ List.of(KeyValue.pair(new Windowed<>(key, windows[0]), 10L)),
+
toListAndCloseIterator(bytesStore.readOnly(IsolationLevel.READ_COMMITTED).fetch(Bytes.wrap(key.getBytes()),
0, windows[0].start())));
+ }
+
+ @ParameterizedTest
+ @MethodSource("getKeySchemas")
+ public void commitShouldMakeStagedWritesVisibleToReadCommitted(final
SegmentedBytesStore.KeySchema schema) {
+ initTransactional(schema);
+ final Bytes storeKey = serializeKey(new Windowed<>("a", windows[0]));
+
+ bytesStore.put(storeKey, serializeValue(50L));
+ // Before commit, READ_COMMITTED cannot see the staged write.
+
assertNull(bytesStore.readOnly(IsolationLevel.READ_COMMITTED).get(storeKey));
+
+ bytesStore.commit(Map.of());
+
+ // After commit, both isolation levels converge on the now-durable
value.
+ assertEquals(50L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_COMMITTED).get(storeKey)));
+ assertEquals(50L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ }
+
+ @ParameterizedTest
+ @MethodSource("getKeySchemas")
+ public void stagedWritesShouldNotAdvanceCommittedPositionUntilCommit(final
SegmentedBytesStore.KeySchema schema) {
+ initTransactional(schema);
+
+ context.setRecordContext(new ProcessorRecordContext(0, 1L, 0, "input",
new RecordHeaders()));
+ bytesStore.put(serializeKey(new Windowed<>("a", windows[0])),
serializeValue(10L));
+ bytesStore.commit(Map.of());
+
+ context.setRecordContext(new ProcessorRecordContext(0, 5L, 0, "input",
new RecordHeaders()));
+ bytesStore.put(serializeKey(new Windowed<>("b", windows[0])),
serializeValue(20L));
+
+ // committed position stays at the last committed offset; the merged
position reflects the staged write.
+ assertEquals(Map.of(0, 1L),
bytesStore.getCommittedPosition().getPartitionPositions("input"));
+ assertEquals(Map.of(0, 5L),
bytesStore.getPosition().getPartitionPositions("input"));
+
+ bytesStore.commit(Map.of());
+ assertEquals(Map.of(0, 5L),
bytesStore.getCommittedPosition().getPartitionPositions("input"));
+ }
+
+ @ParameterizedTest
+ @MethodSource("getKeySchemas")
+ public void approximateNumUncommittedBytesShouldReflectStagedWrites(final
SegmentedBytesStore.KeySchema schema) {
+ initTransactional(schema);
+ assertEquals(0, bytesStore.approximateNumUncommittedBytes());
+
+ bytesStore.put(serializeKey(new Windowed<>("a", windows[0])),
serializeValue(10L));
+ final long staged = bytesStore.approximateNumUncommittedBytes();
+ assertTrue(staged > 0, "a staged write should contribute uncommitted
bytes");
+
+ bytesStore.commit(Map.of());
+ // commit flushes the staged data, so uncommitted bytes drop below the
staged size (a small residual remains, as for any RocksDBStore).
+ assertTrue(bytesStore.approximateNumUncommittedBytes() < staged,
"commit should flush staged writes");
+ }
+
+ @ParameterizedTest
+ @MethodSource("getKeySchemas")
+ public void
nonTransactionalStoreShouldReadIdenticallyAcrossIsolationLevels(final
SegmentedBytesStore.KeySchema schema) {
+ before(schema);
+ final Bytes storeKey = serializeKey(new Windowed<>("a", windows[0]));
+ bytesStore.put(storeKey, serializeValue(10L));
+
+ assertEquals(10L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ assertEquals(10L,
deserializeValue(bytesStore.readOnly(IsolationLevel.READ_COMMITTED).get(storeKey)));
+ }
+
+ private void initTransactional(final SegmentedBytesStore.KeySchema schema)
{
+ before(schema);
+ // Re-open under a transactional EOS context so writes stage until
commit().
+ bytesStore.close();
+ bytesStore = getBytesStore();
+ context = getTransactionalEOSProcessorContext();
+ bytesStore.init(context, bytesStore);
+ }
+
+ private InternalMockProcessorContext<?, ?>
getTransactionalEOSProcessorContext() {
+ final Properties streamsProps = StreamsTestUtils.getStreamsConfig();
+ streamsProps.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+
streamsProps.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ return new InternalMockProcessorContext<>(
+ stateDir,
+ Serdes.String(),
+ Serdes.Long(),
+ new MockRecordCollector(),
+ new ThreadCache(new LogContext("testCache "), 0, new
MockStreamsMetrics(new Metrics())),
+ new StreamsConfig(streamsProps));
+ }
+
+ private Long deserializeValue(final byte[] value) {
+ return new LongDeserializer().deserialize("", value);
+ }
+
private Bytes serializeKey(final Windowed<String> key) {
final StateSerdes<String, Long> stateSerdes =
StateSerdes.withBuiltinTypes("dummy", String.class, Long.class);
if (schema instanceof SessionKeySchema) {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
index 6c3400c344e..24bc67dc00e 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/PlainToHeadersWindowStoreAdapterTest.java
@@ -75,7 +75,7 @@ public class PlainToHeadersWindowStoreAdapterTest {
new StreamsConfig(props)
);
- final SegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
+ final RocksDBSegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
"iqv2-test-store",
"test-metrics-scope",
RETENTION_PERIOD,
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeadersTest.java
index 9acbdfef326..2e04ab05852 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeadersTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBSessionStoreWithHeadersTest.java
@@ -63,7 +63,7 @@ public class RocksDBSessionStoreWithHeadersTest {
new StreamsConfig(props)
);
- final SegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
+ final RocksDBSegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
STORE_NAME,
"test-metrics-scope",
RETENTION_PERIOD,
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeadersTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeadersTest.java
index 554f5559865..04ee23555a7 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeadersTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBTimestampedWindowStoreWithHeadersTest.java
@@ -66,7 +66,7 @@ public class RocksDBTimestampedWindowStoreWithHeadersTest {
new StreamsConfig(props)
);
- final SegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
+ final RocksDBSegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
STORE_NAME,
"test-metrics-scope",
RETENTION_PERIOD,
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
index cf1649b144c..4b757785c42 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/TimestampedToHeadersWindowStoreAdapterTest.java
@@ -76,7 +76,7 @@ public class TimestampedToHeadersWindowStoreAdapterTest {
new StreamsConfig(props)
);
- final SegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
+ final RocksDBSegmentedBytesStore segmentedBytesStore = new
RocksDBSegmentedBytesStore(
"iqv2-test-store",
"test-metrics-scope",
RETENTION_PERIOD,
