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 75e66426bf5 KAFKA-20768: Add transactional and IQ isolation to
time-ordered RocksDB stores (#22829)
75e66426bf5 is described below
commit 75e66426bf533b97898e7c558bd96712dd61c440
Author: Nick Telford <[email protected]>
AuthorDate: Thu Jul 16 00:47:56 2026 +0100
KAFKA-20768: Add transactional and IQ isolation to time-ordered RocksDB
stores (#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)
Reviewers: Bill Bejeck <[email protected]>
---
gradle/spotbugs-exclude.xml | 1 +
...stractDualSchemaRocksDBSegmentedBytesStore.java | 80 ++++++++---
...tractRocksDBTimeOrderedSegmentedBytesStore.java | 158 +++++++++++++++++----
.../RocksDBTimeOrderedKeyValueBytesStore.java | 4 +-
...cksDBTimeOrderedSessionSegmentedBytesStore.java | 31 +++-
.../internals/RocksDBTimeOrderedSessionStore.java | 129 ++++++++++++-----
...ocksDBTimeOrderedWindowSegmentedBytesStore.java | 11 +-
.../internals/RocksDBTimeOrderedWindowStore.java | 125 +++++++++++-----
...ctDualSchemaRocksDBSegmentedBytesStoreTest.java | 116 +++++++++++++++
9 files changed, 528 insertions(+), 127 deletions(-)
diff --git a/gradle/spotbugs-exclude.xml b/gradle/spotbugs-exclude.xml
index 7268ffc6ebe..4c12255637d 100644
--- a/gradle/spotbugs-exclude.xml
+++ b/gradle/spotbugs-exclude.xml
@@ -623,6 +623,7 @@ For a detailed description of spotbugs bug categories, see
https://spotbugs.read
<Class
name="org.apache.kafka.streams.processor.internals.StreamsProducer"/>
<Class
name="org.apache.kafka.streams.processor.internals.TaskManager"/>
<Class
name="org.apache.kafka.streams.state.internals.AbstractRocksDBSegmentedBytesStore"/>
+ <Class
name="org.apache.kafka.streams.state.internals.AbstractDualSchemaRocksDBSegmentedBytesStore"/>
<Class
name="org.apache.kafka.streams.state.internals.InMemoryTimeOrderedKeyValueChangeBuffer"/>
<Class
name="org.apache.kafka.streams.state.internals.RocksDBStore"/>
<Class
name="org.apache.kafka.streams.state.internals.TimeOrderedCachingWindowStore"/>
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.java
index 9dbb4210aa7..a90d87b9077 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStore.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;
@@ -60,6 +61,9 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
protected long observedStreamTime = ConsumerRecord.NO_TIMESTAMP;
protected boolean consistencyEnabled = false;
protected 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;
AbstractDualSchemaRocksDBSegmentedBytesStore(final String name,
@@ -81,36 +85,28 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
@Override
public KeyValueIterator<Bytes, byte[]> all() {
-
- final long actualFrom = getActualFrom(0, baseKeySchema instanceof
PrefixedWindowKeySchemas.TimeFirstWindowKeySchema);
-
- final List<S> searchSpace = segments.allSegments(true);
- final Bytes from = baseKeySchema.lowerRange(null, actualFrom);
- final Bytes to = baseKeySchema.upperRange(null, Long.MAX_VALUE);
-
- return new SegmentIterator<>(
- searchSpace.iterator(),
- baseKeySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, true),
- from,
- to,
- true);
+ return all(true, null);
}
@Override
public KeyValueIterator<Bytes, byte[]> backwardAll() {
+ return all(false, null);
+ }
+ KeyValueIterator<Bytes, byte[]> all(final boolean forward, final
IsolationLevel isolationLevel) {
final long actualFrom = getActualFrom(0, baseKeySchema instanceof
PrefixedWindowKeySchemas.TimeFirstWindowKeySchema);
- final List<S> searchSpace = segments.allSegments(false);
+ final List<S> searchSpace = segments.allSegments(forward);
final Bytes from = baseKeySchema.lowerRange(null, actualFrom);
final Bytes to = baseKeySchema.upperRange(null, Long.MAX_VALUE);
return new SegmentIterator<>(
searchSpace.iterator(),
- baseKeySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, false),
+ baseKeySchema.hasNextCondition(null, null, actualFrom,
Long.MAX_VALUE, forward),
from,
to,
- false);
+ forward,
+ isolationLevel);
}
@Override
@@ -197,7 +193,8 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
LOG.warn("Skipping record for expired segment.");
} 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);
// Put to index first so that if put to base failed, when we
iterate index, we will
// find no base value. If put to base first but putting to
index fails, when we iterate
@@ -214,6 +211,10 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
@Override
public byte[] get(final Bytes rawKey) {
+ return get(rawKey, null);
+ }
+
+ byte[] get(final Bytes rawKey, final IsolationLevel isolationLevel) {
final long timestampFromRawKey =
baseKeySchema.segmentTimestamp(rawKey);
// check if timestamp is expired
@@ -235,7 +236,8 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
if (segment == null) {
return null;
}
- return segment.get(rawKey);
+ // Live read for the vanilla path; interactive queries read through
the segment's isolation view.
+ return isolationLevel == null ? segment.get(rawKey) :
segment.readOnly(isolationLevel).get(rawKey);
}
@Override
@@ -259,6 +261,7 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
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
@@ -275,11 +278,24 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
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")
@@ -293,10 +309,21 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
return segments.committedOffset(partition);
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ long total = 0;
+ for (final S segment : segments.allSegments(true)) {
+ total += segment.approximateNumUncommittedBytes();
+ }
+ return total;
+ }
+
@Override
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
@@ -335,6 +362,21 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStore<S extends Seg
@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;
+ }
+
+ Position getCommittedPosition() {
+ if (transactional) {
+ synchronized (position) {
+ return position.copy();
+ }
+ }
return position;
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBTimeOrderedSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBTimeOrderedSegmentedBytesStore.java
index bec19c91abc..fb0b2741147 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBTimeOrderedSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractRocksDBTimeOrderedSegmentedBytesStore.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.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.errors.ProcessorStateException;
@@ -34,6 +35,7 @@ import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.Objects;
import java.util.Optional;
import java.util.function.Function;
@@ -74,10 +76,13 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
public abstract class IndexToBaseStoreIterator implements
KeyValueIterator<Bytes, byte[]> {
private final KeyValueIterator<Bytes, byte[]> indexIterator;
private byte[] cachedValue;
+ // Null for the owner path (live base reads plus stale-index repair);
non-null for isolation-bound
+ // interactive queries, which read the base store through its
isolation view and never mutate the index.
+ private final IsolationLevel isolationLevel;
-
- IndexToBaseStoreIterator(final KeyValueIterator<Bytes, byte[]>
indexIterator) {
+ IndexToBaseStoreIterator(final KeyValueIterator<Bytes, byte[]>
indexIterator, final IsolationLevel isolationLevel) {
this.indexIterator = indexIterator;
+ this.isolationLevel = isolationLevel;
}
@Override
@@ -99,11 +104,14 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
final Bytes key = indexIterator.peekNextKey();
final Bytes baseKey = getBaseKey(key);
- cachedValue = get(baseKey);
+ cachedValue = get(baseKey, isolationLevel);
if (cachedValue == null) {
// Key not in base store or key is expired, inconsistency
happened and remove from index.
+ // Interactive queries read through an isolation view and
must not mutate, so skip the repair.
indexIterator.next();
-
AbstractRocksDBTimeOrderedSegmentedBytesStore.this.removeIndex(key);
+ if (isolationLevel == null) {
+
AbstractRocksDBTimeOrderedSegmentedBytesStore.this.removeIndex(key);
+ }
} else {
return true;
}
@@ -199,7 +207,8 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
return fetch(key, from, to, false);
}
- protected abstract IndexToBaseStoreIterator
getIndexToBaseStoreIterator(final SegmentIterator<S> segmentIterator);
+ protected abstract IndexToBaseStoreIterator
getIndexToBaseStoreIterator(final SegmentIterator<S> segmentIterator,
+
final IsolationLevel isolationLevel);
public void put(final Bytes key, final long timestamp, final int seqnum,
final byte[] value) {
throw new UnsupportedOperationException("This store does not support
put with timestamp and seqnum");
@@ -229,6 +238,14 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
final long from,
final long to,
final boolean forward) {
+ return fetch(key, from, to, forward, null);
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetch(final Bytes key,
+ final long from,
+ final long to,
+ final boolean forward,
+ final IsolationLevel isolationLevel)
{
final long actualFrom = getActualFrom(from, baseKeySchema instanceof
PrefixedWindowKeySchemas.TimeFirstWindowKeySchema);
@@ -248,7 +265,8 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
indexKeySchema.get().hasNextCondition(key, key, actualFrom,
to, forward),
binaryFrom,
binaryTo,
- forward));
+ forward,
+ isolationLevel), isolationLevel);
}
@@ -263,7 +281,8 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
baseKeySchema.hasNextCondition(key, key, actualFrom, to, forward),
binaryFrom,
binaryTo,
- forward);
+ forward,
+ isolationLevel);
}
@Override
@@ -287,6 +306,15 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
final long from,
final long to,
final boolean forward) {
+ return fetch(keyFrom, keyTo, from, to, forward, null);
+ }
+
+ 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, " +
@@ -313,7 +341,8 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
indexKeySchema.get().hasNextCondition(keyFrom, keyTo,
actualFrom, to, forward),
binaryFrom,
binaryTo,
- forward));
+ forward,
+ isolationLevel), isolationLevel);
}
final List<S> searchSpace = baseKeySchema.segmentsToSearch(segments,
actualFrom, to,
@@ -327,12 +356,26 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
baseKeySchema.hasNextCondition(keyFrom, keyTo, actualFrom, to,
forward),
binaryFrom,
binaryTo,
- forward);
+ forward,
+ isolationLevel);
}
@Override
public KeyValueIterator<Bytes, byte[]> fetchAll(final long timeFrom,
final long timeTo) {
+ return fetchAll(timeFrom, timeTo, true, null);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> backwardFetchAll(final long
timeFrom,
+ final long timeTo)
{
+ return fetchAll(timeFrom, timeTo, false, null);
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetchAll(final long timeFrom,
+ final long timeTo,
+ final boolean forward,
+ final IsolationLevel
isolationLevel) {
final long actualFrom = getActualFrom(timeFrom, baseKeySchema
instanceof PrefixedWindowKeySchemas.TimeFirstWindowKeySchema);
@@ -340,37 +383,98 @@ public abstract class
AbstractRocksDBTimeOrderedSegmentedBytesStore<S extends Se
return KeyValueIterators.emptyIterator();
}
- final List<S> searchSpace = segments.segments(actualFrom, timeTo,
true);
+ final List<S> searchSpace = segments.segments(actualFrom, timeTo,
forward);
final Bytes binaryFrom = baseKeySchema.lowerRange(null, actualFrom);
final Bytes binaryTo = baseKeySchema.upperRange(null, timeTo);
return new SegmentIterator<>(
searchSpace.iterator(),
- baseKeySchema.hasNextCondition(null, null, actualFrom, timeTo,
true),
+ baseKeySchema.hasNextCondition(null, null, actualFrom, timeTo,
forward),
binaryFrom,
binaryTo,
- true);
+ forward,
+ isolationLevel);
}
- @Override
- public KeyValueIterator<Bytes, byte[]> backwardFetchAll(final long
timeFrom,
- final long timeTo)
{
+ /**
+ * As {@link #fetchSession(Bytes, long, long)}, but reading through the
segment isolation view for the
+ * given level. Only the session store supports this; other stores throw.
+ */
+ byte[] fetchSession(final Bytes key,
+ final long sessionStartTime,
+ final long sessionEndTime,
+ final IsolationLevel isolationLevel) {
+ throw new UnsupportedOperationException("This store does not support
fetchSession");
+ }
- final long actualFrom = getActualFrom(timeFrom, baseKeySchema
instanceof PrefixedWindowKeySchemas.TimeFirstWindowKeySchema);
+ /**
+ * As {@link #fetchSessions(long, long)}, but reading through the segment
isolation view for the given
+ * level. Only the session store supports this; other stores throw.
+ */
+ KeyValueIterator<Bytes, byte[]> fetchSessions(final long
earliestSessionEndTime,
+ final long
latestSessionEndTime,
+ final IsolationLevel
isolationLevel) {
+ throw new UnsupportedOperationException("This store does not support
fetchSessions");
+ }
- if (baseKeySchema instanceof
PrefixedWindowKeySchemas.TimeFirstWindowKeySchema && timeTo < actualFrom) {
- return KeyValueIterators.emptyIterator();
+ /** An isolation-bound read view over this store, delegating to the
store's isolation-aware read utilities. */
+ ReadOnlyView readOnly(final IsolationLevel isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(this, isolationLevel);
+ }
+
+ /** Read view bound to an {@link IsolationLevel}; used by the wrapping
window and session stores for IQ reads. */
+ static final class ReadOnlyView {
+ private final AbstractRocksDBTimeOrderedSegmentedBytesStore<?> store;
+ private final IsolationLevel isolationLevel;
+
+ private ReadOnlyView(final
AbstractRocksDBTimeOrderedSegmentedBytesStore<?> store, final IsolationLevel
isolationLevel) {
+ this.store = store;
+ this.isolationLevel = isolationLevel;
}
- final List<S> searchSpace = segments.segments(actualFrom, timeTo,
false);
- final Bytes binaryFrom = baseKeySchema.lowerRange(null, actualFrom);
- final Bytes binaryTo = baseKeySchema.upperRange(null, timeTo);
+ byte[] get(final Bytes rawKey) {
+ return store.get(rawKey, isolationLevel);
+ }
- return new SegmentIterator<>(
- searchSpace.iterator(),
- baseKeySchema.hasNextCondition(null, null, actualFrom, timeTo,
false),
- binaryFrom,
- binaryTo,
- false);
+ 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[]> 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);
+ }
+
+ KeyValueIterator<Bytes, byte[]> all() {
+ return store.all(true, isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> backwardAll() {
+ return store.all(false, isolationLevel);
+ }
+
+ byte[] fetchSession(final Bytes key, final long sessionStartTime,
final long sessionEndTime) {
+ return store.fetchSession(key, sessionStartTime, sessionEndTime,
isolationLevel);
+ }
+
+ KeyValueIterator<Bytes, byte[]> fetchSessions(final long
earliestSessionEndTime, final long latestSessionEndTime) {
+ return store.fetchSessions(earliestSessionEndTime,
latestSessionEndTime, isolationLevel);
+ }
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
index 37c6dce156b..e8de482e160 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedKeyValueBytesStore.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.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import
org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.TimeFirstWindowKeySchema;
@@ -57,7 +58,8 @@ public class RocksDBTimeOrderedKeyValueBytesStore extends
AbstractRocksDBTimeOrd
}
@Override
- protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<KeyValueSegment> segmentIterator) {
+ protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<KeyValueSegment> segmentIterator,
+ final
IsolationLevel isolationLevel) {
throw new UnsupportedOperationException("Do not use for
TimeOrderedKeyValueStore");
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java
index dbb7713ab35..40dfdfea071 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionSegmentedBytesStore.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.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.kstream.Window;
@@ -38,8 +39,9 @@ import java.util.Optional;
public class RocksDBTimeOrderedSessionSegmentedBytesStore<S extends Segment>
extends AbstractRocksDBTimeOrderedSegmentedBytesStore<S> {
private class SessionKeySchemaIndexToBaseStoreIterator extends
IndexToBaseStoreIterator {
- SessionKeySchemaIndexToBaseStoreIterator(final KeyValueIterator<Bytes,
byte[]> indexIterator) {
- super(indexIterator);
+ SessionKeySchemaIndexToBaseStoreIterator(final KeyValueIterator<Bytes,
byte[]> indexIterator,
+ final IsolationLevel
isolationLevel) {
+ super(indexIterator, isolationLevel);
}
@Override
@@ -67,16 +69,31 @@ public class RocksDBTimeOrderedSessionSegmentedBytesStore<S
extends Segment> ext
public byte[] fetchSession(final Bytes key,
final long sessionStartTime,
final long sessionEndTime) {
+ return fetchSession(key, sessionStartTime, sessionEndTime, null);
+ }
+
+ @Override
+ byte[] fetchSession(final Bytes key,
+ final long sessionStartTime,
+ final long sessionEndTime,
+ final IsolationLevel isolationLevel) {
return get(TimeFirstSessionKeySchema.toBinary(
key,
sessionStartTime,
sessionEndTime
- ));
+ ), isolationLevel);
}
@Override
public KeyValueIterator<Bytes, byte[]> fetchSessions(final long
earliestSessionEndTime,
final long
latestSessionEndTime) {
+ return fetchSessions(earliestSessionEndTime, latestSessionEndTime,
null);
+ }
+
+ @Override
+ KeyValueIterator<Bytes, byte[]> fetchSessions(final long
earliestSessionEndTime,
+ final long
latestSessionEndTime,
+ final IsolationLevel
isolationLevel) {
final List<S> searchSpace = segments.segments(earliestSessionEndTime,
latestSessionEndTime, true);
// here we want [0, latestSE, FF] as the upper bound to cover any
possible keys,
@@ -102,7 +119,8 @@ public class RocksDBTimeOrderedSessionSegmentedBytesStore<S
extends Segment> ext
},
binaryFrom,
binaryTo,
- true);
+ true,
+ isolationLevel);
}
@Override
@@ -133,7 +151,8 @@ public class RocksDBTimeOrderedSessionSegmentedBytesStore<S
extends Segment> ext
}
@Override
- protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<S> segmentIterator) {
- return new SessionKeySchemaIndexToBaseStoreIterator(segmentIterator);
+ protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<S> segmentIterator,
+ final
IsolationLevel isolationLevel) {
+ return new SessionKeySchemaIndexToBaseStoreIterator(segmentIterator,
isolationLevel);
}
}
\ No newline at end of file
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStore.java
index ab375907f5d..8fd0b736718 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedSessionStore.java
@@ -16,18 +16,22 @@
*/
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
org.apache.kafka.streams.state.internals.PrefixedSessionKeySchemas.TimeFirstSessionKeySchema;
+import java.time.Instant;
import java.util.Objects;
public class RocksDBTimeOrderedSessionStore
@@ -52,12 +56,15 @@ public class RocksDBTimeOrderedSessionStore
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
);
}
@@ -65,32 +72,21 @@ public class RocksDBTimeOrderedSessionStore
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final long
earliestSessionEndTime,
final long
latestSessionEndTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().fetchSessions(earliestSessionEndTime, latestSessionEndTime);
- return new WrappedSessionStoreIterator(bytesIterator,
TimeFirstSessionKeySchema::from);
+ return sessionIterator(wrapped().fetchSessions(earliestSessionEndTime,
latestSessionEndTime));
}
@Override
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,
TimeFirstSessionKeySchema::from);
+ 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,
TimeFirstSessionKeySchema::from);
+ return sessionIterator(wrapped().backwardFetch(key,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
@@ -98,13 +94,7 @@ public class RocksDBTimeOrderedSessionStore
final Bytes
keyTo,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator = wrapped().fetch(
- keyFrom,
- keyTo,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator,
TimeFirstSessionKeySchema::from);
+ return sessionIterator(wrapped().fetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
@@ -112,24 +102,19 @@ public class RocksDBTimeOrderedSessionStore
final Bytes keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetch(
- keyFrom,
- keyTo,
- earliestSessionEndTime,
- latestSessionStartTime
- );
- return new WrappedSessionStoreIterator(bytesIterator,
TimeFirstSessionKeySchema::from);
+ return sessionIterator(wrapped().backwardFetch(keyFrom, keyTo,
earliestSessionEndTime, latestSessionStartTime));
}
@Override
public byte[] fetchSession(final Bytes key,
final long sessionStartTime,
final long sessionEndTime) {
- return wrapped().fetchSession(
- key,
- sessionStartTime,
- sessionEndTime
- );
+ return wrapped().fetchSession(key, sessionStartTime, sessionEndTime);
+ }
+
+ // 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,
TimeFirstSessionKeySchema::from);
}
@Override
@@ -161,4 +146,80 @@ public class RocksDBTimeOrderedSessionStore
public void put(final Windowed<Bytes> sessionKey, final byte[] aggregate) {
wrapped().put(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
AbstractRocksDBTimeOrderedSegmentedBytesStore.ReadOnlyView segmented;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.segmented = wrapped().readOnly(isolationLevel);
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final long
sessionStartTime, final long sessionEndTime) {
+ return segmented.fetchSession(key, sessionStartTime,
sessionEndTime);
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final Instant
sessionStartTime, final Instant sessionEndTime) {
+ return fetchSession(key, sessionStartTime.toEpochMilli(),
sessionEndTime.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/RocksDBTimeOrderedWindowSegmentedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowSegmentedBytesStore.java
index 249933df1ff..f69562524a9 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowSegmentedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowSegmentedBytesStore.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.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.state.KeyValueIterator;
@@ -47,8 +48,9 @@ public class RocksDBTimeOrderedWindowSegmentedBytesStore<S
extends Segment> exte
* This can be reused by both window store implementations (with and
without headers).
*/
class WindowKeySchemaIndexToBaseStoreIterator extends
IndexToBaseStoreIterator {
- WindowKeySchemaIndexToBaseStoreIterator(final KeyValueIterator<Bytes,
byte[]> indexIterator) {
- super(indexIterator);
+ WindowKeySchemaIndexToBaseStoreIterator(final KeyValueIterator<Bytes,
byte[]> indexIterator,
+ final IsolationLevel
isolationLevel) {
+ super(indexIterator, isolationLevel);
}
@Override
@@ -103,7 +105,8 @@ public class RocksDBTimeOrderedWindowSegmentedBytesStore<S
extends Segment> exte
}
@Override
- protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<S> segmentIterator) {
- return new WindowKeySchemaIndexToBaseStoreIterator(segmentIterator);
+ protected IndexToBaseStoreIterator getIndexToBaseStoreIterator(final
SegmentIterator<S> segmentIterator,
+ final
IsolationLevel isolationLevel) {
+ return new WindowKeySchemaIndexToBaseStoreIterator(segmentIterator,
isolationLevel);
}
}
\ No newline at end of file
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowStore.java
index ebe1f4c5614..deec8374762 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTimeOrderedWindowStore.java
@@ -16,21 +16,25 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
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.TimestampedBytesStore;
import org.apache.kafka.streams.state.WindowStore;
import org.apache.kafka.streams.state.WindowStoreIterator;
import
org.apache.kafka.streams.state.internals.PrefixedWindowKeySchemas.TimeFirstWindowKeySchema;
+import java.time.Instant;
import java.util.Map;
import java.util.Objects;
@@ -98,20 +102,12 @@ public class RocksDBTimeOrderedWindowStore<S extends
Segment>
@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,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).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,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).valuesIterator();
+ return valuesIterator(wrapped().backwardFetch(key, timeFrom, timeTo));
}
@Override
@@ -119,11 +115,7 @@ public class RocksDBTimeOrderedWindowStore<S extends
Segment>
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,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).keyValueIterator();
+ return keyValueIterator(wrapped().fetch(keyFrom, keyTo, timeFrom,
timeTo));
}
@Override
@@ -131,43 +123,38 @@ public class RocksDBTimeOrderedWindowStore<S extends
Segment>
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,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).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,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).keyValueIterator();
+ return keyValueIterator(wrapped().all());
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardAll();
- return new WindowStoreIteratorWrapper(bytesIterator,
- windowSize,
- TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).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 keyValueIterator(wrapped().fetchAll(timeFrom, timeTo));
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final
long timeFrom, final long 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,
TimeFirstWindowKeySchema::extractStoreTimestamp,
- TimeFirstWindowKeySchema::fromStoreBytesKey).keyValueIterator();
+ TimeFirstWindowKeySchema::fromStoreBytesKey).valuesIterator();
}
- @Override
- public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetchAll(final
long timeFrom, final long timeTo) {
- final KeyValueIterator<Bytes, byte[]> bytesIterator =
wrapped().backwardFetchAll(timeFrom, timeTo);
+ private KeyValueIterator<Windowed<Bytes>, byte[]> keyValueIterator(final
KeyValueIterator<Bytes, byte[]> bytesIterator) {
return new WindowStoreIteratorWrapper(bytesIterator,
windowSize,
TimeFirstWindowKeySchema::extractStoreTimestamp,
@@ -178,17 +165,83 @@ public class RocksDBTimeOrderedWindowStore<S extends
Segment>
return wrapped().hasIndex();
}
+ @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
AbstractRocksDBTimeOrderedSegmentedBytesStore.ReadOnlyView segmented;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.segmented = wrapped().readOnly(isolationLevel);
+ }
+
+ @Override
+ public byte[] fetch(final Bytes key, final long time) {
+ return
segmented.get(TimeFirstWindowKeySchema.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/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java
index 9d2ac218f85..3e51e20a35d 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/AbstractDualSchemaRocksDBSegmentedBytesStoreTest.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;
@@ -25,6 +26,7 @@ import org.apache.kafka.common.header.internals.RecordHeaders;
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.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
@@ -1682,6 +1684,120 @@ public abstract class
AbstractDualSchemaRocksDBSegmentedBytesStoreTest {
return Set.of(Objects.requireNonNull(windowDir.list()));
}
+ @Test
+ public void readCommittedShouldHideStagedWritesUntilCommit() {
+ initTransactional();
+ 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(timeOrdered().readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ assertEquals(10L,
deserializeValue(timeOrdered().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(timeOrdered().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(timeOrdered().readOnly(IsolationLevel.READ_COMMITTED).fetch(Bytes.wrap(key.getBytes()),
0, windows[0].start())));
+ }
+
+ @Test
+ public void commitShouldMakeStagedWritesVisibleToReadCommitted() {
+ initTransactional();
+ final Bytes storeKey = serializeKey(new Windowed<>("a", windows[0]));
+
+ bytesStore.put(storeKey, serializeValue(50L));
+ // Before commit, READ_COMMITTED cannot see the staged write.
+
assertNull(timeOrdered().readOnly(IsolationLevel.READ_COMMITTED).get(storeKey));
+
+ bytesStore.commit(Map.of());
+
+ // After commit, both isolation levels converge on the now-durable
value.
+ assertEquals(50L,
deserializeValue(timeOrdered().readOnly(IsolationLevel.READ_COMMITTED).get(storeKey)));
+ assertEquals(50L,
deserializeValue(timeOrdered().readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ }
+
+ @Test
+ public void stagedWritesShouldNotAdvanceCommittedPositionUntilCommit() {
+ initTransactional();
+
+ 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"));
+ }
+
+ @Test
+ public void approximateNumUncommittedBytesShouldReflectStagedWrites() {
+ initTransactional();
+ 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");
+ }
+
+ @Test
+ public void
nonTransactionalStoreShouldReadIdenticallyAcrossIsolationLevels() {
+ // before() already initialised the store under a non-transactional
context.
+ final Bytes storeKey = serializeKey(new Windowed<>("a", windows[0]));
+ bytesStore.put(storeKey, serializeValue(10L));
+
+ assertEquals(10L,
deserializeValue(timeOrdered().readOnly(IsolationLevel.READ_UNCOMMITTED).get(storeKey)));
+ assertEquals(10L,
deserializeValue(timeOrdered().readOnly(IsolationLevel.READ_COMMITTED).get(storeKey)));
+ }
+
+ @SuppressWarnings("unchecked")
+ private AbstractRocksDBTimeOrderedSegmentedBytesStore<KeyValueSegment>
timeOrdered() {
+ return
(AbstractRocksDBTimeOrderedSegmentedBytesStore<KeyValueSegment>) bytesStore;
+ }
+
+ private void initTransactional() {
+ // 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) {
return serializeKey(key, false);
}
