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 a884fadfa4d KAFKA-20499: Add isolation-level reads to in-memory stores
(#22681)
a884fadfa4d is described below
commit a884fadfa4d1fe8b1c0e2f14a46ccf9491fd1f70
Author: Nick Telford <[email protected]>
AuthorDate: Tue Jun 30 23:26:51 2026 +0100
KAFKA-20499: Add isolation-level reads to in-memory stores (#22681)
Part of the KIP-892 interactive-query isolation-level series. With
transactional state stores enabled, the in-memory key-value, window, and
session stores buffer uncommitted writes in a transaction buffer, and
their reads consulted that buffer unconditionally. An interactive query
therefore had no way to request only committed data — a `READ_COMMITTED`
query would observe writes still staged in the open transaction.
This overrides `readOnly(IsolationLevel)` on each of the three in-memory
stores to return a view that consults the transaction buffer under
`READ_UNCOMMITTED` and bypasses it under `READ_COMMITTED`, reading the
base map directly. For the window and session stores the buffer dispatch
in each read path is extracted into private helpers parameterised by the
buffer (collapsing the forward/backward pairs into shared helpers), so
the view can opt in or out without duplicating the iteration logic. The
key-value view synchronises on the store's monitor so concurrent IQ
reads observe a consistent snapshot relative to processor-thread writes.
Semantic tests across the three store families assert that
`READ_UNCOMMITTED` observes staged writes (and deletes) while
`READ_COMMITTED` sees only the last committed snapshot.
This branched off the now-merged in-memory transactional-store and
IQ-isolation-framework work and now applies directly to trunk.
Reviewers: Bill Bejeck <[email protected]>
---
.../state/internals/AbstractTransactionBuffer.java | 52 ++++-
.../state/internals/InMemoryKeyValueStore.java | 141 +++++++++---
.../state/internals/InMemorySessionStore.java | 245 ++++++++++++++-------
.../InMemorySessionTransactionBuffer.java | 36 ++-
.../state/internals/InMemoryTransactionBuffer.java | 21 ++
.../state/internals/InMemoryWindowStore.java | 174 +++++++++++----
.../internals/InMemoryWindowTransactionBuffer.java | 32 ++-
.../ChangeLoggingKeyValueBytesStoreTest.java | 21 +-
.../state/internals/InMemoryKeyValueStoreTest.java | 176 +++++++++++++++
.../state/internals/InMemorySessionStoreTest.java | 150 +++++++++++++
.../state/internals/InMemoryWindowStoreTest.java | 171 ++++++++++++++
11 files changed, 1056 insertions(+), 163 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionBuffer.java
index 865fe9fb07e..95881cb8022 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionBuffer.java
@@ -16,6 +16,8 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
+
import java.util.NavigableMap;
import java.util.Optional;
import java.util.TreeMap;
@@ -124,14 +126,22 @@ abstract class AbstractTransactionBuffer<K extends
Comparable<K>> implements Tra
@Override
public ManagedKeyValueIterator<K, byte[]> all(final boolean forward) {
+ return all(forward, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ /**
+ * As {@link #all(boolean)}, but under {@link
IsolationLevel#READ_COMMITTED} the staging layer is
+ * excluded and only the committed base store is iterated.
+ */
+ ManagedKeyValueIterator<K, byte[]> all(final boolean forward, final
IsolationLevel isolationLevel) {
if (Thread.currentThread() == ownerThread) {
// Eager copy of the staging layer (lock-free for the sole
mutator) so a later owner
// stage cannot invalidate this iterator; the base store is
iterated live.
- final NavigableMap<K, Optional<byte[]>> stagingSnapshot = new
TreeMap<>(pendingWrites);
+ final NavigableMap<K, Optional<byte[]>> stagingSnapshot =
stagingSnapshot(null, null, true, isolationLevel);
final ManagedKeyValueIterator<K, byte[]> baseIter =
newBaseIterator(null, null, forward, true);
return new StagedMergeIterator<>(stagingSnapshot, baseIter,
forward);
}
- return snapshotScan(null, null, forward, true);
+ return snapshotScan(null, null, forward, true, isolationLevel);
}
/**
@@ -139,12 +149,21 @@ abstract class AbstractTransactionBuffer<K extends
Comparable<K>> implements Tra
*/
@Override
public ManagedKeyValueIterator<K, byte[]> range(final K from, final K to,
final boolean forward, final boolean toInclusive) {
+ return range(from, to, forward, toInclusive,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ /**
+ * As {@link #range(Object, Object, boolean, boolean)}, but under {@link
IsolationLevel#READ_COMMITTED}
+ * the staging layer is excluded and only the committed base store is
iterated.
+ */
+ ManagedKeyValueIterator<K, byte[]> range(final K from, final K to, final
boolean forward, final boolean toInclusive,
+ final IsolationLevel
isolationLevel) {
if (Thread.currentThread() == ownerThread) {
- final NavigableMap<K, Optional<byte[]>> stagingSnapshot = new
TreeMap<>(boundStaging(from, to, toInclusive));
+ final NavigableMap<K, Optional<byte[]>> stagingSnapshot =
stagingSnapshot(from, to, toInclusive, isolationLevel);
final ManagedKeyValueIterator<K, byte[]> baseIter =
newBaseIterator(from, to, forward, toInclusive);
return new StagedMergeIterator<>(stagingSnapshot, baseIter,
forward);
}
- return snapshotScan(from, to, forward, toInclusive);
+ return snapshotScan(from, to, forward, toInclusive, isolationLevel);
}
@Override
@@ -195,9 +214,20 @@ abstract class AbstractTransactionBuffer<K extends
Comparable<K>> implements Tra
*/
ManagedKeyValueIterator<K, byte[]> snapshotScan(final K from, final K to,
final boolean forward,
final boolean toInclusive) {
+ return snapshotScan(from, to, forward, toInclusive,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ /**
+ * As {@link #snapshotScan(Object, Object, boolean, boolean)}, but under
+ * {@link IsolationLevel#READ_COMMITTED} only the committed base store is
snapshotted, excluding the
+ * staging layer.
+ */
+ ManagedKeyValueIterator<K, byte[]> snapshotScan(final K from, final K to,
+ final boolean forward,
final boolean toInclusive,
+ final IsolationLevel
isolationLevel) {
snapshotLock.readLock().lock();
try {
- final NavigableMap<K, Optional<byte[]>> stagingSnapshot = new
TreeMap<>(boundStaging(from, to, toInclusive));
+ final NavigableMap<K, Optional<byte[]>> stagingSnapshot =
stagingSnapshot(from, to, toInclusive, isolationLevel);
final ManagedKeyValueIterator<K, byte[]> baseIter =
newBaseSnapshotIterator(from, to, forward, toInclusive);
return new StagedMergeIterator<>(stagingSnapshot, baseIter,
forward);
} finally {
@@ -205,6 +235,18 @@ abstract class AbstractTransactionBuffer<K extends
Comparable<K>> implements Tra
}
}
+ /**
+ * An eager copy of the bounded staging range, so a later owner {@code
stage} cannot invalidate an
+ * open iterator. Under {@link IsolationLevel#READ_COMMITTED} the copy is
empty, which makes the
+ * {@link StagedMergeIterator} degenerate to a committed-only (base) view.
+ */
+ private NavigableMap<K, Optional<byte[]>> stagingSnapshot(final K from,
final K to, final boolean toInclusive,
+ final
IsolationLevel isolationLevel) {
+ return isolationLevel == IsolationLevel.READ_UNCOMMITTED
+ ? new TreeMap<>(boundStaging(from, to, toInclusive))
+ : new TreeMap<>();
+ }
+
/**
* Constructs a view over the transaction buffer's read buffer consisting
of only the keys in the given range.
* If {@code from} or {@code to} are {@code null}, the range will be
open-ended; if they are both {@code null}, the
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
index ae020cf0c08..0a1f94ef321 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.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.serialization.Serializer;
import org.apache.kafka.common.utils.Bytes;
@@ -34,6 +35,7 @@ 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.KeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -43,6 +45,7 @@ import java.util.List;
import java.util.Map;
import java.util.NavigableMap;
import java.util.NoSuchElementException;
+import java.util.Objects;
import java.util.Set;
import java.util.TreeMap;
import java.util.TreeSet;
@@ -141,14 +144,23 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
}
@Override
- public synchronized byte[] get(final Bytes key) {
+ public byte[] get(final Bytes key) {
+ return get(key, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private byte[] get(final Bytes key, final IsolationLevel isolationLevel) {
if (transactionBuffer != null) {
- final java.util.Optional<byte[]> staged =
transactionBuffer.get(key);
- if (staged != null) {
- return staged.orElse(null);
+ if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
+ final java.util.Optional<byte[]> staged =
transactionBuffer.get(key);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
}
+ return transactionBuffer.getCommitted(key);
+ }
+ synchronized (this) {
+ return map.get(key);
}
- return map.get(key);
}
@Override
@@ -196,15 +208,21 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
}
@Override
- public synchronized <PS extends Serializer<P>, P> KeyValueIterator<Bytes,
byte[]> prefixScan(final P prefix, final PS prefixKeySerializer) {
+ public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]>
prefixScan(final P prefix, final PS prefixKeySerializer) {
+ return prefixScan(prefix, prefixKeySerializer,
IsolationLevel.READ_UNCOMMITTED);
+ }
+ private <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]>
prefixScan(final P prefix, final PS prefixKeySerializer,
+
final IsolationLevel isolationLevel) {
final Bytes from = Bytes.wrap(prefixKeySerializer.serialize(null,
prefix));
final Bytes to = ByteUtils.increment(from);
if (transactionBuffer != null) {
- return transactionBuffer.range(from, to, true, false);
+ return transactionBuffer.range(from, to, true, false,
isolationLevel);
+ }
+ synchronized (this) {
+ return new InMemoryKeyValueIterator(map.subMap(from, true, to,
false).keySet(), true);
}
- return new InMemoryKeyValueIterator(map.subMap(from, true, to,
false).keySet(), true);
}
@Override
@@ -218,33 +236,37 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
}
@Override
- public synchronized KeyValueIterator<Bytes, byte[]> range(final Bytes
from, final Bytes to) {
- return range(from, to, true);
+ public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final Bytes
to) {
+ return range(from, to, true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
- public synchronized KeyValueIterator<Bytes, byte[]> reverseRange(final
Bytes from, final Bytes to) {
- return range(from, to, false);
+ public KeyValueIterator<Bytes, byte[]> reverseRange(final Bytes from,
final Bytes to) {
+ return range(from, to, false, IsolationLevel.READ_UNCOMMITTED);
}
- private KeyValueIterator<Bytes, byte[]> range(final Bytes from, final
Bytes to, final boolean forward) {
- if (transactionBuffer != null) {
- return transactionBuffer.range(from, to, forward, true);
- }
- if (from == null && to == null) {
- return getKeyValueIterator(map.keySet(), forward);
- } else if (from == null) {
- return getKeyValueIterator(map.headMap(to, true).keySet(),
forward);
- } else if (to == null) {
- return getKeyValueIterator(map.tailMap(from, true).keySet(),
forward);
- } else if (from.compareTo(to) > 0) {
+ private KeyValueIterator<Bytes, byte[]> range(final Bytes from, final
Bytes to, final boolean forward,
+ final IsolationLevel
isolationLevel) {
+ if (from != null && to != null && from.compareTo(to) > 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, " +
"or serdes that don't preserve ordering when
lexicographically comparing the serialized bytes. " +
"Note that the built-in numerical serdes do not follow
this for negative numbers");
return KeyValueIterators.emptyIterator();
- } else {
- return getKeyValueIterator(map.subMap(from, true, to,
true).keySet(), forward);
+ }
+ if (transactionBuffer != null) {
+ return transactionBuffer.range(from, to, forward, true,
isolationLevel);
+ }
+ synchronized (this) {
+ if (from == null && to == null) {
+ return getKeyValueIterator(map.keySet(), forward);
+ } else if (from == null) {
+ return getKeyValueIterator(map.headMap(to, true).keySet(),
forward);
+ } else if (to == null) {
+ return getKeyValueIterator(map.tailMap(from, true).keySet(),
forward);
+ } else {
+ return getKeyValueIterator(map.subMap(from, true, to,
true).keySet(), forward);
+ }
}
}
@@ -253,13 +275,13 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
}
@Override
- public synchronized KeyValueIterator<Bytes, byte[]> all() {
- return range(null, null);
+ public KeyValueIterator<Bytes, byte[]> all() {
+ return range(null, null, true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
- public synchronized KeyValueIterator<Bytes, byte[]> reverseAll() {
- return new InMemoryKeyValueIterator(map.keySet(), false);
+ public KeyValueIterator<Bytes, byte[]> reverseAll() {
+ return range(null, null, false, IsolationLevel.READ_UNCOMMITTED);
}
@Override
@@ -267,6 +289,67 @@ public class InMemoryKeyValueStore implements
KeyValueStore<Bytes, byte[]> {
return map.size();
}
+ @Override
+ public ReadOnlyKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /**
+ * Read-only view of this store bound to an isolation level. Every read
delegates to the store's
+ * shared read helper with this view's {@code isolationLevel}, so the view
and the store's own
+ * reads follow one code path; under READ_COMMITTED that path excludes the
transaction buffer's
+ * staging layer and snapshots the committed base under the buffer's
read-lock, so an interactive
+ * query cannot race a concurrent commit.
+ */
+ private final class ReadOnlyView implements ReadOnlyKeyValueStore<Bytes,
byte[]> {
+
+ private final IsolationLevel isolationLevel;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.isolationLevel = isolationLevel;
+ }
+
+ @Override
+ public byte[] get(final Bytes key) {
+ Objects.requireNonNull(key, "key cannot be null");
+ return InMemoryKeyValueStore.this.get(key, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final
Bytes to) {
+ return InMemoryKeyValueStore.this.range(from, to, true,
isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseRange(final Bytes from,
final Bytes to) {
+ return InMemoryKeyValueStore.this.range(from, to, false,
isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> all() {
+ return InMemoryKeyValueStore.this.range(null, null, true,
isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseAll() {
+ return InMemoryKeyValueStore.this.range(null, null, false,
isolationLevel);
+ }
+
+ @Override
+ public <PS extends Serializer<P>, P> KeyValueIterator<Bytes, byte[]>
prefixScan(final P prefix,
+
final PS prefixKeySerializer) {
+ Objects.requireNonNull(prefix, "prefix cannot be null");
+ Objects.requireNonNull(prefixKeySerializer, "prefixKeySerializer
cannot be null");
+ return InMemoryKeyValueStore.this.prefixScan(prefix,
prefixKeySerializer, isolationLevel);
+ }
+
+ @Override
+ public long approximateNumEntries() {
+ return InMemoryKeyValueStore.this.approximateNumEntries();
+ }
+ }
+
@Override
public long approximateNumUncommittedBytes() {
if (transactionBuffer != null) {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
index b44389772bc..0b36f5b11c4 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionStore.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;
@@ -37,11 +38,13 @@ 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.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import java.time.Instant;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
@@ -252,6 +255,13 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
public byte[] fetchSession(final Bytes key,
final long sessionStartTime,
final long sessionEndTime) {
+ return fetchSession(key, sessionStartTime, sessionEndTime,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private byte[] fetchSession(final Bytes key,
+ final long sessionStartTime,
+ final long sessionEndTime,
+ final IsolationLevel isolationLevel) {
removeExpiredSegments();
Objects.requireNonNull(key, "key cannot be null");
@@ -259,10 +269,15 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
// Only need to search if the record hasn't expired yet
if (sessionEndTime > observedStreamTime - retentionPeriod) {
if (transactionBuffer != null) {
- final Optional<byte[]> staged = transactionBuffer.get(key,
sessionStartTime, sessionEndTime);
- if (staged != null) {
- return staged.orElse(null);
+ if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
+ final Optional<byte[]> staged = transactionBuffer.get(key,
sessionStartTime, sessionEndTime);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
}
+ // Committed read of the base map, taken under the buffer's
snapshot read-lock so it
+ // cannot race a concurrent commit.
+ return transactionBuffer.getCommitted(key, sessionStartTime,
sessionEndTime);
}
final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyMap = endTimeMap.get(sessionEndTime);
@@ -279,11 +294,17 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final long
earliestSessionEndTime,
final long
latestSessionEndTime) {
+ return findSessions(earliestSessionEndTime, latestSessionEndTime,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final long
earliestSessionEndTime,
+ final long
latestSessionEndTime,
+ final
IsolationLevel isolationLevel) {
removeExpiredSegments();
if (transactionBuffer != null) {
return newTransactionalSessionIterator(
- null, null, Long.MAX_VALUE, earliestSessionEndTime,
latestSessionEndTime, true
+ null, null, Long.MAX_VALUE, earliestSessionEndTime,
latestSessionEndTime, true, isolationLevel
);
}
@@ -297,44 +318,38 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final Bytes
key,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- Objects.requireNonNull(key, "key cannot be null");
-
- removeExpiredSegments();
-
- if (transactionBuffer != null) {
- return newTransactionalSessionIterator(
- key, key, latestSessionStartTime, earliestSessionEndTime,
Long.MAX_VALUE, true
- );
- }
-
- return registerNewIterator(key,
- key,
- latestSessionStartTime,
- endTimeMap.tailMap(earliestSessionEndTime,
true).entrySet().iterator(),
- true);
+ return findSessionsForKey(key, earliestSessionEndTime,
latestSessionStartTime, true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes key,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
+ return findSessionsForKey(key, earliestSessionEndTime,
latestSessionStartTime, false, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> findSessionsForKey(final
Bytes key,
+ final
long earliestSessionEndTime,
+ final
long latestSessionStartTime,
+ final
boolean forward,
+ final
IsolationLevel isolationLevel) {
Objects.requireNonNull(key, "key cannot be null");
removeExpiredSegments();
if (transactionBuffer != null) {
return newTransactionalSessionIterator(
- key, key, latestSessionStartTime, earliestSessionEndTime,
Long.MAX_VALUE, false
+ key, key, latestSessionStartTime, earliestSessionEndTime,
Long.MAX_VALUE, forward, isolationLevel
);
}
- return registerNewIterator(
- key,
- key,
- latestSessionStartTime,
- endTimeMap.tailMap(earliestSessionEndTime,
true).descendingMap().entrySet().iterator(),
- false
- );
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeSubMap =
+ endTimeMap.tailMap(earliestSessionEndTime, true);
+ final Iterator<Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>>> endTimeIter =
+ forward
+ ? endTimeSubMap.entrySet().iterator()
+ : endTimeSubMap.descendingMap().entrySet().iterator();
+ return registerNewIterator(key, key, latestSessionStartTime,
endTimeIter, forward);
}
@Override
@@ -342,24 +357,7 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final Bytes
keyTo,
final long
earliestSessionEndTime,
final long
latestSessionStartTime) {
- removeExpiredSegments();
-
- if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
- LOG.warn(INVALID_RANGE_WARN_MSG);
- return KeyValueIterators.emptyIterator();
- }
-
- if (transactionBuffer != null) {
- return newTransactionalSessionIterator(
- keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, Long.MAX_VALUE, true
- );
- }
-
- return registerNewIterator(keyFrom,
- keyTo,
- latestSessionStartTime,
- endTimeMap.tailMap(earliestSessionEndTime,
true).entrySet().iterator(),
- true);
+ return findSessionsForKeyRange(keyFrom, keyTo, earliestSessionEndTime,
latestSessionStartTime, true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
@@ -367,6 +365,15 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final Bytes keyTo,
final long earliestSessionEndTime,
final long latestSessionStartTime) {
+ return findSessionsForKeyRange(keyFrom, keyTo, earliestSessionEndTime,
latestSessionStartTime, false, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]>
findSessionsForKeyRange(final Bytes keyFrom,
+
final Bytes keyTo,
+
final long earliestSessionEndTime,
+
final long latestSessionStartTime,
+
final boolean forward,
+
final IsolationLevel isolationLevel) {
removeExpiredSegments();
if (keyFrom != null && keyTo != null && keyFrom.compareTo(keyTo) > 0) {
@@ -376,68 +383,68 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
if (transactionBuffer != null) {
return newTransactionalSessionIterator(
- keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, Long.MAX_VALUE, false
+ keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, Long.MAX_VALUE, forward, isolationLevel
);
}
- return registerNewIterator(
- keyFrom,
- keyTo,
- latestSessionStartTime,
- endTimeMap.tailMap(earliestSessionEndTime,
true).descendingMap().entrySet().iterator(),
- false
- );
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeSubMap =
+ endTimeMap.tailMap(earliestSessionEndTime, true);
+ final Iterator<Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>>> endTimeIter =
+ forward
+ ? endTimeSubMap.entrySet().iterator()
+ : endTimeSubMap.descendingMap().entrySet().iterator();
+ return registerNewIterator(keyFrom, keyTo, latestSessionStartTime,
endTimeIter, forward);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes key) {
-
- Objects.requireNonNull(key, "key cannot be null");
-
- removeExpiredSegments();
-
- if (transactionBuffer != null) {
- return newTransactionalSessionIterator(key, key, Long.MAX_VALUE,
0, Long.MAX_VALUE, true);
- }
-
- return registerNewIterator(key, key, Long.MAX_VALUE,
endTimeMap.entrySet().iterator(), true);
+ return fetchForKey(key, true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final Bytes
key) {
+ return fetchForKey(key, false, IsolationLevel.READ_UNCOMMITTED);
+ }
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetchForKey(final Bytes
key,
+ final
boolean forward,
+ final
IsolationLevel isolationLevel) {
Objects.requireNonNull(key, "key cannot be null");
removeExpiredSegments();
if (transactionBuffer != null) {
- return newTransactionalSessionIterator(key, key, Long.MAX_VALUE,
0, Long.MAX_VALUE, false);
+ return newTransactionalSessionIterator(key, key, Long.MAX_VALUE,
0, Long.MAX_VALUE, forward, isolationLevel);
}
- return registerNewIterator(key, key, Long.MAX_VALUE,
endTimeMap.descendingMap().entrySet().iterator(), false);
+ final Iterator<Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>>> endTimeIter =
+ forward ? endTimeMap.entrySet().iterator() :
endTimeMap.descendingMap().entrySet().iterator();
+ return registerNewIterator(key, key, Long.MAX_VALUE, endTimeIter,
forward);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo) {
- removeExpiredSegments();
-
- if (transactionBuffer != null) {
- return newTransactionalSessionIterator(keyFrom, keyTo,
Long.MAX_VALUE, 0, Long.MAX_VALUE, true);
- }
-
- return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE,
endTimeMap.entrySet().iterator(), true);
+ return fetchForKeyRange(keyFrom, keyTo, true,
IsolationLevel.READ_UNCOMMITTED);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final Bytes
keyFrom, final Bytes keyTo) {
+ return fetchForKeyRange(keyFrom, keyTo, false,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetchForKeyRange(final
Bytes keyFrom,
+ final
Bytes keyTo,
+ final
boolean forward,
+ final
IsolationLevel isolationLevel) {
removeExpiredSegments();
if (transactionBuffer != null) {
- return newTransactionalSessionIterator(keyFrom, keyTo,
Long.MAX_VALUE, 0, Long.MAX_VALUE, false);
+ return newTransactionalSessionIterator(keyFrom, keyTo,
Long.MAX_VALUE, 0, Long.MAX_VALUE, forward, isolationLevel);
}
- return registerNewIterator(
- keyFrom, keyTo, Long.MAX_VALUE,
endTimeMap.descendingMap().entrySet().iterator(), false);
+ final Iterator<Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>>> endTimeIter =
+ forward ? endTimeMap.entrySet().iterator() :
endTimeMap.descendingMap().entrySet().iterator();
+ return registerNewIterator(keyFrom, keyTo, Long.MAX_VALUE,
endTimeIter, forward);
}
@Override
@@ -465,6 +472,86 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
);
}
+ @Override
+ public ReadOnlySessionStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /**
+ * Read-only view of this store. For a transactional store the {@code
isolationLevel} is passed
+ * down to the transactional read path (which excludes the staging layer
under READ_COMMITTED), so
+ * both isolation levels share the one path and the read is snapshotted
under the buffer's
+ * read-lock — an interactive query cannot race a concurrent commit. For a
non-transactional store
+ * reads hit {@code endTimeMap} directly.
+ */
+ private final class ReadOnlyView implements ReadOnlySessionStore<Bytes,
byte[]> {
+
+ private final IsolationLevel isolationLevel;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.isolationLevel = isolationLevel;
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final long startTime,
final long endTime) {
+ return InMemorySessionStore.this.fetchSession(key, startTime,
endTime, isolationLevel);
+ }
+
+ @Override
+ public byte[] fetchSession(final Bytes key, final Instant startTime,
final Instant endTime) {
+ return fetchSession(key, startTime.toEpochMilli(),
endTime.toEpochMilli());
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final
Bytes key,
+ final
long earliestSessionEndTime,
+ final
long latestSessionStartTime) {
+ return InMemorySessionStore.this.findSessionsForKey(key,
earliestSessionEndTime, latestSessionStartTime, true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes key,
+
final long earliestSessionEndTime,
+
final long latestSessionStartTime) {
+ return InMemorySessionStore.this.findSessionsForKey(key,
earliestSessionEndTime, latestSessionStartTime, false, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> findSessions(final
Bytes keyFrom, final Bytes keyTo,
+ final
long earliestSessionEndTime,
+ final
long latestSessionStartTime) {
+ return InMemorySessionStore.this.findSessionsForKeyRange(keyFrom,
keyTo, earliestSessionEndTime, latestSessionStartTime, true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFindSessions(final Bytes keyFrom, final Bytes keyTo,
+
final long earliestSessionEndTime,
+
final long latestSessionStartTime) {
+ return InMemorySessionStore.this.findSessionsForKeyRange(keyFrom,
keyTo, earliestSessionEndTime, latestSessionStartTime, false, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
key) {
+ return InMemorySessionStore.this.fetchForKey(key, true,
isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes key) {
+ return InMemorySessionStore.this.fetchForKey(key, false,
isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo) {
+ return InMemorySessionStore.this.fetchForKeyRange(keyFrom, keyTo,
true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes keyFrom, final Bytes keyTo) {
+ return InMemorySessionStore.this.fetchForKeyRange(keyFrom, keyTo,
false, isolationLevel);
+ }
+ }
+
@Override
public long approximateNumUncommittedBytes() {
if (transactionBuffer != null) {
@@ -526,11 +613,12 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final long latestSessionStartTime,
final long earliestSessionEndTime,
final long latestSessionEndTime,
- final boolean forward) {
+ final boolean forward,
+ final IsolationLevel isolationLevel) {
final TransactionalSessionIterator iterator = new
TransactionalSessionIterator(
transactionBuffer, keyFrom, keyTo, latestSessionStartTime,
earliestSessionEndTime, latestSessionEndTime,
- observedStreamTime - retentionPeriod, forward,
openTransactionalIterators::remove
+ observedStreamTime - retentionPeriod, forward, isolationLevel,
openTransactionalIterators::remove
);
openTransactionalIterators.add(iterator);
return iterator;
@@ -578,6 +666,7 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final long latestSessionEndTime,
final long oldestRetainedEndTime,
final boolean forward,
+ final IsolationLevel isolationLevel,
final Consumer<TransactionalSessionIterator> deregister) {
this.keyFrom = keyFrom;
this.keyTo = keyTo;
@@ -592,7 +681,7 @@ public class InMemorySessionStore implements
SessionStore<Bytes, byte[]>, WithRe
final InMemorySessionTransactionBuffer.SessionEntryKey to =
new
InMemorySessionTransactionBuffer.SessionEntryKey(latestSessionEndTime, keyTo,
0);
- this.delegate = buffer.range(from, to, forward, true);
+ this.delegate = buffer.range(from, to, forward, true,
isolationLevel);
}
@Override
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
index 00e794a7fcb..00d4e8246fd 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemorySessionTransactionBuffer.java
@@ -160,6 +160,39 @@ class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemor
timeRange = endTimeMap;
}
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> copy = deepCopy(timeRange);
+ return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ }
+
+ /**
+ * Committed-only point read for a single session, bypassing the staging
layer. Non-owner (IQ)
+ * reads take the snapshot read-lock so the read reflects a single
committed state rather than a
+ * commit in progress.
+ */
+ byte[] getCommitted(final Bytes key, final long startTime, final long
endTime) {
+ if (Thread.currentThread() == ownerThread) {
+ return baseGet(key, startTime, endTime);
+ }
+ snapshotLock.readLock().lock();
+ try {
+ return baseGet(key, startTime, endTime);
+ } finally {
+ snapshotLock.readLock().unlock();
+ }
+ }
+
+ private byte[] baseGet(final Bytes key, final long startTime, final long
endTime) {
+ final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyMap = endTimeMap.get(endTime);
+ if (keyMap == null) {
+ return null;
+ }
+ final ConcurrentNavigableMap<Long, byte[]> startTimeMap =
keyMap.get(key);
+ return startTimeMap == null ? null : startTimeMap.get(startTime);
+ }
+
+ /** Deep-copies a bounded end-time range into a private map so iterators
are isolated from later mutation. */
+ private static ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> deepCopy(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> timeRange) {
final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> copy = new ConcurrentSkipListMap<>();
for (final Map.Entry<Long, ConcurrentNavigableMap<Bytes,
ConcurrentNavigableMap<Long, byte[]>>> endTimeEntry : timeRange.entrySet()) {
final ConcurrentNavigableMap<Bytes, ConcurrentNavigableMap<Long,
byte[]>> keyCopy = new ConcurrentSkipListMap<>();
@@ -168,8 +201,7 @@ class InMemorySessionTransactionBuffer extends
AbstractTransactionBuffer<InMemor
}
copy.put(endTimeEntry.getKey(), keyCopy);
}
-
- return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ return copy;
}
/**
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
index b1b2f319f5c..26a8ef8f61c 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryTransactionBuffer.java
@@ -81,6 +81,27 @@ class InMemoryTransactionBuffer extends
AbstractTransactionBuffer<Bytes> {
return new BaseMapIterator(forward ? copy : copy.descendingMap());
}
+ /**
+ * Committed-only point read. Bypasses the staging layer and reads the
base map directly.
+ * <p>
+ * The owner reads lock-free (it is the sole mutator and single-threaded).
Non-owner (IQ) reads
+ * take the snapshot read-lock: {@code commit}/{@code rollback} mutate the
non-thread-safe base
+ * {@link java.util.TreeMap} under the write-lock, so an unlocked read
could traverse a
+ * mid-rebalance tree (wrong value/NPE) or observe a partially-applied
commit. The read-lock also
+ * supplies the happens-before edge that makes committed writes visible.
+ */
+ byte[] getCommitted(final Bytes key) {
+ if (Thread.currentThread() == ownerThread) {
+ return baseMap.get(key);
+ }
+ snapshotLock.readLock().lock();
+ try {
+ return baseMap.get(key);
+ } finally {
+ snapshotLock.readLock().unlock();
+ }
+ }
+
private NavigableMap<Bytes, byte[]> boundView(final Bytes from, final
Bytes to, final boolean toInclusive) {
if (from != null && to != null) {
return baseMap.subMap(from, true, to, toInclusive);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
index 26f9634326b..a9bdc43cc50 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowStore.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;
@@ -38,6 +39,7 @@ 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;
@@ -45,6 +47,7 @@ import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.nio.ByteBuffer;
+import java.time.Instant;
import java.util.Iterator;
import java.util.Map;
import java.util.NoSuchElementException;
@@ -221,6 +224,10 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
@Override
public byte[] fetch(final Bytes key, final long windowStartTimestamp) {
+ return fetch(key, windowStartTimestamp,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private byte[] fetch(final Bytes key, final long windowStartTimestamp,
final IsolationLevel isolationLevel) {
Objects.requireNonNull(key, "key cannot be null");
removeExpiredSegments();
@@ -230,10 +237,15 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
}
if (transactionBuffer != null) {
- final Optional<byte[]> staged =
transactionBuffer.get(windowStartTimestamp, key);
- if (staged != null) {
- return staged.orElse(null);
+ if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
+ final Optional<byte[]> staged =
transactionBuffer.get(windowStartTimestamp, key);
+ if (staged != null) {
+ return staged.orElse(null);
+ }
}
+ // Committed read of the base map, taken under the buffer's
snapshot read-lock so it
+ // cannot race a concurrent commit.
+ return transactionBuffer.getCommitted(windowStartTimestamp, key);
}
final ConcurrentNavigableMap<Bytes, byte[]> kvMap =
segmentMap.get(windowStartTimestamp);
@@ -255,6 +267,11 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
}
WindowStoreIterator<byte[]> fetch(final Bytes key, final long timeFrom,
final long timeTo, final boolean forward) {
+ return fetch(key, timeFrom, timeTo, forward,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private WindowStoreIterator<byte[]> fetch(final Bytes key, final long
timeFrom, final long timeTo,
+ final boolean forward, final
IsolationLevel isolationLevel) {
Objects.requireNonNull(key, "key cannot be null");
removeExpiredSegments();
@@ -270,7 +287,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final Bytes keyFrom = retainDuplicates ? wrapForDups(key, 0) : key;
final Bytes keyTo = retainDuplicates ? wrapForDups(key,
Integer.MAX_VALUE) : key;
return registerTransactional(new TransactionalWindowStoreIterator(
- transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates,
+ transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates, isolationLevel,
openTransactionalIterators::remove
));
}
@@ -278,15 +295,13 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
if (forward) {
return registerNewWindowStoreIterator(
key,
- segmentMap.subMap(minTime, true, timeTo, true)
- .entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).entrySet().iterator(),
true
);
} else {
return registerNewWindowStoreIterator(
key,
- segmentMap.subMap(minTime, true, timeTo, true)
- .descendingMap().entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).descendingMap().entrySet().iterator(),
false
);
}
@@ -313,6 +328,15 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final long timeFrom,
final long timeTo,
final boolean forward) {
+ return fetch(from, to, timeFrom, timeTo, forward,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes from,
+ final Bytes to,
+ final long
timeFrom,
+ final long timeTo,
+ final boolean
forward,
+ final
IsolationLevel isolationLevel) {
removeExpiredSegments();
if (from != null && to != null && from.compareTo(to) > 0) {
@@ -334,7 +358,8 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final Bytes keyFrom = (retainDuplicates && from != null) ?
wrapForDups(from, 0) : from;
final Bytes keyTo = (retainDuplicates && to != null) ?
wrapForDups(to, Integer.MAX_VALUE) : to;
return registerTransactional(new
TransactionalWindowedKeyValueIterator(
- transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates, windowSize, openTransactionalIterators::remove
+ transactionBuffer, keyFrom, keyTo, minTime, timeTo, forward,
retainDuplicates, windowSize, isolationLevel,
+ openTransactionalIterators::remove
));
}
@@ -342,16 +367,14 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return registerNewWindowedKeyValueIterator(
from,
to,
- segmentMap.subMap(minTime, true, timeTo, true)
- .entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).entrySet().iterator(),
true
);
} else {
return registerNewWindowedKeyValueIterator(
from,
to,
- segmentMap.subMap(minTime, true, timeTo, true)
- .descendingMap().entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).descendingMap().entrySet().iterator(),
false
);
}
@@ -368,6 +391,12 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
}
KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long timeFrom,
final long timeTo, final boolean forward) {
+ return fetchAll(timeFrom, timeTo, forward,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final long
timeFrom, final long timeTo,
+ final boolean
forward,
+ final
IsolationLevel isolationLevel) {
removeExpiredSegments();
// add one b/c records expire exactly retentionPeriod ms after created
@@ -379,7 +408,8 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
if (transactionBuffer != null) {
return registerTransactional(new
TransactionalWindowedKeyValueIterator(
- transactionBuffer, null, null, minTime, timeTo, forward,
retainDuplicates, windowSize, openTransactionalIterators::remove
+ transactionBuffer, null, null, minTime, timeTo, forward,
retainDuplicates, windowSize, isolationLevel,
+ openTransactionalIterators::remove
));
}
@@ -387,16 +417,14 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
return registerNewWindowedKeyValueIterator(
null,
null,
- segmentMap.subMap(minTime, true, timeTo, true)
- .entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).entrySet().iterator(),
true
);
} else {
return registerNewWindowedKeyValueIterator(
null,
null,
- segmentMap.subMap(minTime, true, timeTo, true)
- .descendingMap().entrySet().iterator(),
+ segmentMap.subMap(minTime, true, timeTo,
true).descendingMap().entrySet().iterator(),
false
);
}
@@ -404,42 +432,31 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
- removeExpiredSegments();
-
- final long minTime = observedStreamTime - retentionPeriod;
-
- if (transactionBuffer != null) {
- return registerTransactional(new
TransactionalWindowedKeyValueIterator(
- transactionBuffer, null, null, minTime + 1, Long.MAX_VALUE,
true, retainDuplicates, windowSize, openTransactionalIterators::remove
- ));
- }
-
- return registerNewWindowedKeyValueIterator(
- null,
- null,
- segmentMap.tailMap(minTime, false).entrySet().iterator(),
- true
- );
+ return all(true, IsolationLevel.READ_UNCOMMITTED);
}
@Override
public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
+ return all(false, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private KeyValueIterator<Windowed<Bytes>, byte[]> all(final boolean
forward,
+ final IsolationLevel
isolationLevel) {
removeExpiredSegments();
final long minTime = observedStreamTime - retentionPeriod;
if (transactionBuffer != null) {
return registerTransactional(new
TransactionalWindowedKeyValueIterator(
- transactionBuffer, null, null, minTime + 1, Long.MAX_VALUE,
false, retainDuplicates, windowSize, openTransactionalIterators::remove
+ transactionBuffer, null, null, minTime + 1, Long.MAX_VALUE,
forward, retainDuplicates, windowSize, isolationLevel,
+ openTransactionalIterators::remove
));
}
- return registerNewWindowedKeyValueIterator(
- null,
- null,
- segmentMap.tailMap(minTime,
false).descendingMap().entrySet().iterator(),
- false
- );
+ final Iterator<Map.Entry<Long, ConcurrentNavigableMap<Bytes, byte[]>>>
segIter = forward
+ ? segmentMap.tailMap(minTime, false).entrySet().iterator()
+ : segmentMap.tailMap(minTime,
false).descendingMap().entrySet().iterator();
+ return registerNewWindowedKeyValueIterator(null, null, segIter,
forward);
}
@Override
@@ -467,6 +484,75 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
);
}
+ @Override
+ public ReadOnlyWindowStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /**
+ * Read-only view of this store. For a transactional store the {@code
isolationLevel} is passed
+ * down to the transactional read path (which excludes the staging layer
under READ_COMMITTED), so
+ * both isolation levels share the one path and the read is snapshotted
under the buffer's
+ * read-lock — an interactive query cannot race a concurrent commit. For a
non-transactional store
+ * reads hit {@code segmentMap} directly.
+ */
+ private final class ReadOnlyView implements ReadOnlyWindowStore<Bytes,
byte[]> {
+
+ private final IsolationLevel isolationLevel;
+
+ ReadOnlyView(final IsolationLevel isolationLevel) {
+ this.isolationLevel = isolationLevel;
+ }
+
+ @Override
+ public byte[] fetch(final Bytes key, final long time) {
+ return InMemoryWindowStore.this.fetch(key, time, isolationLevel);
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> fetch(final Bytes key, final
Instant timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetch(key,
timeFrom.toEpochMilli(), timeTo.toEpochMilli(), true, isolationLevel);
+ }
+
+ @Override
+ public WindowStoreIterator<byte[]> backwardFetch(final Bytes key,
final Instant timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetch(key,
timeFrom.toEpochMilli(), timeTo.toEpochMilli(), false, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetch(final Bytes
keyFrom, final Bytes keyTo,
+ final Instant
timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetch(keyFrom, keyTo,
timeFrom.toEpochMilli(), timeTo.toEpochMilli(), true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardFetch(final
Bytes keyFrom, final Bytes keyTo,
+ final
Instant timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetch(keyFrom, keyTo,
timeFrom.toEpochMilli(), timeTo.toEpochMilli(), false, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> fetchAll(final
Instant timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetchAll(timeFrom.toEpochMilli(),
timeTo.toEpochMilli(), true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]>
backwardFetchAll(final Instant timeFrom, final Instant timeTo) {
+ return InMemoryWindowStore.this.fetchAll(timeFrom.toEpochMilli(),
timeTo.toEpochMilli(), false, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> all() {
+ return InMemoryWindowStore.this.all(true, isolationLevel);
+ }
+
+ @Override
+ public KeyValueIterator<Windowed<Bytes>, byte[]> backwardAll() {
+ return InMemoryWindowStore.this.all(false, isolationLevel);
+ }
+ }
+
@Override
public long approximateNumUncommittedBytes() {
if (transactionBuffer != null) {
@@ -624,6 +710,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final long timeTo,
final boolean forward,
final boolean retainDuplicates,
+ final IsolationLevel isolationLevel,
final Consumer<KeyValueIterator<?, ?>> deregister) {
this.retainDuplicates = retainDuplicates;
this.deregister = deregister;
@@ -634,7 +721,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final InMemoryWindowTransactionBuffer.WindowEntryKey to =
new InMemoryWindowTransactionBuffer.WindowEntryKey(timeTo,
keyTo);
- this.delegate = buffer.range(from, to, forward, true);
+ this.delegate = buffer.range(from, to, forward, true,
isolationLevel);
}
@Override
@@ -712,6 +799,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final boolean forward,
final boolean retainDuplicates,
final long windowSize,
+ final IsolationLevel isolationLevel,
final Consumer<KeyValueIterator<?, ?>> deregister) {
this.retainDuplicates = retainDuplicates;
this.windowSize = windowSize;
@@ -723,7 +811,7 @@ public class InMemoryWindowStore implements
WindowStore<Bytes, byte[]>, WithRete
final InMemoryWindowTransactionBuffer.WindowEntryKey to =
new InMemoryWindowTransactionBuffer.WindowEntryKey(timeTo,
keyTo);
- this.delegate = buffer.range(from, to, forward, true);
+ this.delegate = buffer.range(from, to, forward, true,
isolationLevel);
}
@Override
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
index 1fd2ac9f753..03fe8ca35b7 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryWindowTransactionBuffer.java
@@ -151,12 +151,40 @@ class InMemoryWindowTransactionBuffer extends
AbstractTransactionBuffer<InMemory
timeRange = segmentMap;
}
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> copy = deepCopy(timeRange);
+ return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ }
+
+ /**
+ * Committed-only point read for a single window, bypassing the staging
layer. Non-owner (IQ)
+ * reads take the snapshot read-lock so the read reflects a single
committed state rather than a
+ * commit in progress.
+ */
+ byte[] getCommitted(final long timestamp, final Bytes key) {
+ if (Thread.currentThread() == ownerThread) {
+ return baseGet(timestamp, key);
+ }
+ snapshotLock.readLock().lock();
+ try {
+ return baseGet(timestamp, key);
+ } finally {
+ snapshotLock.readLock().unlock();
+ }
+ }
+
+ private byte[] baseGet(final long timestamp, final Bytes key) {
+ final ConcurrentNavigableMap<Bytes, byte[]> kvMap =
segmentMap.get(timestamp);
+ return kvMap == null ? null : kvMap.get(key);
+ }
+
+ /** Deep-copies a bounded segment range into a private map so iterators
are isolated from later mutation. */
+ private static ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> deepCopy(
+ final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> timeRange) {
final ConcurrentNavigableMap<Long, ConcurrentNavigableMap<Bytes,
byte[]>> copy = new ConcurrentSkipListMap<>();
for (final Map.Entry<Long, ConcurrentNavigableMap<Bytes, byte[]>>
segment : timeRange.entrySet()) {
copy.put(segment.getKey(), new
ConcurrentSkipListMap<>(segment.getValue()));
}
-
- return baseIterator(forward ? copy : copy.descendingMap(), from, to,
forward);
+ return copy;
}
/**
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
index 65ffd0cffba..76bd6579297 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/ChangeLoggingKeyValueBytesStoreTest.java
@@ -38,6 +38,7 @@ import
org.apache.kafka.streams.processor.internals.metrics.StreamsMetricsImpl;
import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.test.InternalMockProcessorContext;
import org.apache.kafka.test.MockRecordCollector;
import org.apache.kafka.test.TestUtils;
@@ -267,15 +268,27 @@ public class ChangeLoggingKeyValueBytesStoreTest {
}
@Test
+ @SuppressWarnings("unchecked")
public void shouldDelegateReadOnlyUncommittedToInner() {
- assertThat(store.readOnly(IsolationLevel.READ_UNCOMMITTED),
- sameInstance(inner.readOnly(IsolationLevel.READ_UNCOMMITTED)));
+ final KeyValueStore<Bytes, byte[]> innerMock =
mock(KeyValueStore.class);
+ final ChangeLoggingKeyValueBytesStore outer = new
ChangeLoggingKeyValueBytesStore(innerMock);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> view =
mock(ReadOnlyKeyValueStore.class);
+
when(innerMock.readOnly(IsolationLevel.READ_UNCOMMITTED)).thenReturn(view);
+
+ assertThat(outer.readOnly(IsolationLevel.READ_UNCOMMITTED),
sameInstance(view));
+ verify(innerMock).readOnly(IsolationLevel.READ_UNCOMMITTED);
}
@Test
+ @SuppressWarnings("unchecked")
public void shouldDelegateReadOnlyCommittedToInner() {
- assertThat(store.readOnly(IsolationLevel.READ_COMMITTED),
- sameInstance(inner.readOnly(IsolationLevel.READ_COMMITTED)));
+ final KeyValueStore<Bytes, byte[]> innerMock =
mock(KeyValueStore.class);
+ final ChangeLoggingKeyValueBytesStore outer = new
ChangeLoggingKeyValueBytesStore(innerMock);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> view =
mock(ReadOnlyKeyValueStore.class);
+
when(innerMock.readOnly(IsolationLevel.READ_COMMITTED)).thenReturn(view);
+
+ assertThat(outer.readOnly(IsolationLevel.READ_COMMITTED),
sameInstance(view));
+ verify(innerMock).readOnly(IsolationLevel.READ_COMMITTED);
}
private StreamsConfig streamsConfigMock() {
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
index e8b0bb31b38..894d88674fc 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStoreTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
@@ -24,14 +25,19 @@ import
org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.serialization.UUIDSerializer;
import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.KeyValue;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.processor.StateStoreContext;
import org.apache.kafka.streams.processor.internals.ProcessorRecordContext;
import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.state.KeyValueIterator;
import org.apache.kafka.streams.state.KeyValueStore;
import org.apache.kafka.streams.state.KeyValueStoreTestDriver;
+import org.apache.kafka.streams.state.ReadOnlyKeyValueStore;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.BeforeEach;
@@ -41,7 +47,11 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.NoSuchElementException;
+import java.util.Properties;
import java.util.UUID;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
@@ -439,6 +449,172 @@ public class InMemoryKeyValueStoreTest extends
AbstractKeyValueStoreTest {
assertThrows(IllegalStateException.class, iter::peekNextKey);
}
+ @Test
+ public void readOnlyCommittedShouldHideStagedWritesFromTransactionBuffer()
{
+ final InMemoryKeyValueStore txnStore = openTransactionalStore();
+ try {
+ final Bytes k1 = bytesKey("k1");
+ final Bytes k2 = bytesKey("k2");
+ txnStore.put(k1, bytesValue("v1"));
+ txnStore.commit(Map.of());
+
+ txnStore.put(k1, bytesValue("v1-staged"));
+ txnStore.put(k2, bytesValue("v2-staged"));
+
+ final ReadOnlyKeyValueStore<Bytes, byte[]> uncommitted =
txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> committed =
txnStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ assertArrayEquals(bytesValue("v1-staged"), uncommitted.get(k1));
+ assertArrayEquals(bytesValue("v2-staged"), uncommitted.get(k2));
+ assertArrayEquals(bytesValue("v1"), committed.get(k1));
+ assertThat(committed.get(k2), nullValue());
+
+ try (KeyValueIterator<Bytes, byte[]> it = committed.all()) {
+ final List<String> keys = new ArrayList<>();
+ while (it.hasNext()) {
+ keys.add(new String(it.next().key.get()));
+ }
+ assertEquals(List.of("k1"), keys);
+ }
+ try (KeyValueIterator<Bytes, byte[]> it = uncommitted.all()) {
+ final List<String> keys = new ArrayList<>();
+ while (it.hasNext()) {
+ keys.add(new String(it.next().key.get()));
+ }
+ assertEquals(List.of("k1", "k2"), keys);
+ }
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyCommittedShouldNotSeeStagedDelete() {
+ final InMemoryKeyValueStore txnStore = openTransactionalStore();
+ try {
+ final Bytes k = bytesKey("k");
+ txnStore.put(k, bytesValue("v"));
+ txnStore.commit(Map.of());
+
+ txnStore.delete(k);
+
+
assertThat(txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(k),
nullValue());
+ assertArrayEquals(bytesValue("v"),
txnStore.readOnly(IsolationLevel.READ_COMMITTED).get(k));
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyPrefixScanShouldRespectIsolationLevel() {
+ final InMemoryKeyValueStore txnStore = openTransactionalStore();
+ try {
+ txnStore.put(bytesKey("p-1"), bytesValue("a"));
+ txnStore.commit(Map.of());
+ txnStore.put(bytesKey("p-2"), bytesValue("b"));
+ txnStore.put(bytesKey("q-1"), bytesValue("z"));
+
+ final List<String> uncommittedKeys = new ArrayList<>();
+ try (KeyValueIterator<Bytes, byte[]> it =
txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED)
+ .prefixScan("p-", stringSerializer)) {
+ while (it.hasNext()) {
+ uncommittedKeys.add(new String(it.next().key.get()));
+ }
+ }
+ final List<String> committedKeys = new ArrayList<>();
+ try (KeyValueIterator<Bytes, byte[]> it =
txnStore.readOnly(IsolationLevel.READ_COMMITTED)
+ .prefixScan("p-", stringSerializer)) {
+ while (it.hasNext()) {
+ committedKeys.add(new String(it.next().key.get()));
+ }
+ }
+ assertEquals(List.of("p-1", "p-2"), uncommittedKeys);
+ assertEquals(List.of("p-1"), committedKeys);
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyCommittedScanShouldNotRaceConcurrentCommit() throws
InterruptedException {
+ // A READ_COMMITTED scan from a non-owner (IQ) thread must observe a
point-in-time snapshot
+ // of the committed base map, taken under the buffer's snapshot
read-lock. Before the fix the
+ // view copied the live TreeMap under the store monitor, which does
not exclude commit()'s
+ // write-lock-guarded flushToBase(), so a concurrent commit threw
ConcurrentModificationException
+ // and/or exposed a half-applied commit. Each commit flips every key
to a single generation, so
+ // a correct snapshot scan must see one uniform generation.
+ final InMemoryKeyValueStore txnStore = openTransactionalStore();
+ final int numKeys = 500;
+ final int rounds = 300;
+ final AtomicReference<Throwable> readerFailure = new
AtomicReference<>();
+ final AtomicBoolean stop = new AtomicBoolean(false);
+
+ try {
+ for (int i = 0; i < numKeys; i++) {
+ txnStore.put(bytesKey("k" + i), bytesValue("gen0"));
+ }
+ txnStore.commit(Map.of()); // owner thread owns the buffer
+
+ final Thread reader = new Thread(() -> {
+ final ReadOnlyKeyValueStore<Bytes, byte[]> committed =
txnStore.readOnly(IsolationLevel.READ_COMMITTED);
+ try {
+ while (!stop.get()) {
+ try (KeyValueIterator<Bytes, byte[]> it =
committed.all()) {
+ String generation = null;
+ int count = 0;
+ while (it.hasNext()) {
+ final String value = new
String(it.next().value);
+ if (generation == null) {
+ generation = value;
+ } else {
+ assertEquals(generation, value, "scan
observed a partial commit");
+ }
+ count++;
+ }
+ // a committed snapshot is always the full,
uniform key set
+ assertEquals(numKeys, count, "scan observed a
partial commit");
+ }
+ }
+ } catch (final Throwable t) {
+ readerFailure.set(t);
+ }
+ }, "iq-reader");
+ reader.start();
+
+ for (int round = 1; round <= rounds && readerFailure.get() ==
null; round++) {
+ final String generation = "gen" + round;
+ for (int i = 0; i < numKeys; i++) {
+ txnStore.put(bytesKey("k" + i), bytesValue(generation));
+ }
+ txnStore.commit(Map.of());
+ }
+ stop.set(true);
+ reader.join(TimeUnit.SECONDS.toMillis(30));
+
+ if (readerFailure.get() != null) {
+ throw new AssertionError("READ_COMMITTED scan raced a
concurrent commit", readerFailure.get());
+ }
+ } finally {
+ stop.set(true);
+ txnStore.close();
+ }
+ }
+
+ private InMemoryKeyValueStore openTransactionalStore() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(props)
+ );
+ final InMemoryKeyValueStore store = new
InMemoryKeyValueStore("txn-in-memory-store");
+ store.init(ctx, store);
+ return store;
+ }
+
private byte[] bytesValue(final String value) {
return value.getBytes();
}
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
index eca2fa4502f..cfcf156253c 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemorySessionStoreTest.java
@@ -16,18 +16,32 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
+import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.utils.Bytes;
+import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.Windowed;
import org.apache.kafka.streams.kstream.internals.SessionWindow;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlySessionStore;
+import org.apache.kafka.test.InternalMockProcessorContext;
+import org.apache.kafka.test.StreamsTestUtils;
+import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Test;
+import java.util.Map;
+import java.util.Properties;
import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static org.apache.kafka.test.StreamsTestUtils.valuesToSet;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
public class InMemorySessionStoreTest extends AbstractSessionBytesStoreTest {
@@ -120,4 +134,140 @@ public class InMemorySessionStoreTest extends
AbstractSessionBytesStoreTest {
}
}
+ @Test
+ public void readOnlyCommittedShouldHideStagedSessions() {
+ final InMemorySessionStore store = openTransactionalSessionStore();
+ try {
+ final Bytes k = Bytes.wrap("k".getBytes());
+ store.put(new Windowed<>(k, new SessionWindow(0, 10)),
"v1".getBytes());
+ store.commit(Map.of());
+
+ store.put(new Windowed<>(k, new SessionWindow(20, 30)),
"v2".getBytes());
+ store.remove(new Windowed<>(k, new SessionWindow(0, 10)));
+
+ final ReadOnlySessionStore<Bytes, byte[]> uncommitted =
store.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlySessionStore<Bytes, byte[]> committed =
store.readOnly(IsolationLevel.READ_COMMITTED);
+
+ assertNull(uncommitted.fetchSession(k, 0, 10));
+ assertArrayEquals("v2".getBytes(), uncommitted.fetchSession(k, 20,
30));
+ assertArrayEquals("v1".getBytes(), committed.fetchSession(k, 0,
10));
+ assertNull(committed.fetchSession(k, 20, 30));
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void readOnlyFindSessionsShouldRespectIsolationLevel() {
+ final InMemorySessionStore store = openTransactionalSessionStore();
+ try {
+ final Bytes k = Bytes.wrap("k".getBytes());
+ store.put(new Windowed<>(k, new SessionWindow(0, 10)),
"v1".getBytes());
+ store.commit(Map.of());
+ store.put(new Windowed<>(k, new SessionWindow(20, 30)),
"v2".getBytes());
+
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+
store.readOnly(IsolationLevel.READ_UNCOMMITTED).findSessions(k, 0L, 100L)) {
+ int count = 0;
+ while (it.hasNext()) {
+ it.next();
+ count++;
+ }
+ assertEquals(2, count);
+ }
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+
store.readOnly(IsolationLevel.READ_COMMITTED).findSessions(k, 0L, 100L)) {
+ int count = 0;
+ while (it.hasNext()) {
+ it.next();
+ count++;
+ }
+ assertEquals(1, count);
+ }
+ } finally {
+ store.close();
+ }
+ }
+
+ @Test
+ public void readOnlyCommittedFindSessionsShouldNotRaceConcurrentCommit()
throws InterruptedException {
+ // A READ_COMMITTED scan from a non-owner (IQ) thread must observe a
point-in-time snapshot of
+ // the committed sessions, taken under the buffer's snapshot
read-lock. Without it the view
+ // iterated the live end-time map while commit()'s write-lock-guarded
flushToBase() mutated it,
+ // exposing a half-applied commit. Each commit flips every session to
a single generation, so a
+ // correct snapshot scan must see one uniform generation across the
full session set.
+ final InMemorySessionStore store = openTransactionalSessionStore();
+ final Bytes k = Bytes.wrap("k".getBytes());
+ final int numSessions = 50;
+ final int rounds = 300;
+ final AtomicReference<Throwable> readerFailure = new
AtomicReference<>();
+ final AtomicBoolean stop = new AtomicBoolean(false);
+
+ try {
+ for (int i = 0; i < numSessions; i++) {
+ store.put(new Windowed<>(k, new SessionWindow(i * 100L, i *
100L + 10L)), "gen0".getBytes());
+ }
+ store.commit(Map.of());
+
+ final Thread reader = new Thread(() -> {
+ final ReadOnlySessionStore<Bytes, byte[]> committed =
store.readOnly(IsolationLevel.READ_COMMITTED);
+ try {
+ while (!stop.get()) {
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+ committed.findSessions(k, 0L, numSessions *
100L)) {
+ String generation = null;
+ int count = 0;
+ while (it.hasNext()) {
+ final String value = new
String(it.next().value);
+ if (generation == null) {
+ generation = value;
+ } else {
+ assertEquals(generation, value, "scan
observed a partial commit");
+ }
+ count++;
+ }
+ assertEquals(numSessions, count, "scan observed a
partial commit");
+ }
+ }
+ } catch (final Throwable t) {
+ readerFailure.set(t);
+ }
+ }, "iq-reader");
+ reader.start();
+
+ for (int round = 1; round <= rounds && readerFailure.get() ==
null; round++) {
+ final String generation = "gen" + round;
+ for (int i = 0; i < numSessions; i++) {
+ store.put(new Windowed<>(k, new SessionWindow(i * 100L, i
* 100L + 10L)), generation.getBytes());
+ }
+ store.commit(Map.of());
+ }
+ stop.set(true);
+ reader.join(TimeUnit.SECONDS.toMillis(30));
+
+ if (readerFailure.get() != null) {
+ throw new AssertionError("READ_COMMITTED findSessions raced a
concurrent commit", readerFailure.get());
+ }
+ } finally {
+ stop.set(true);
+ store.close();
+ }
+ }
+
+ private InMemorySessionStore openTransactionalSessionStore() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(props)
+ );
+ final InMemorySessionStore store = new InMemorySessionStore(
+ "txn-in-memory-session-store", RETENTION_PERIOD, "scope");
+ store.init(ctx, store);
+ return store;
+ }
+
}
\ No newline at end of file
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
index b9461f75d05..be6981ca9f5 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/InMemoryWindowStoreTest.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes;
@@ -31,6 +32,7 @@ import org.apache.kafka.streams.query.QueryResult;
import org.apache.kafka.streams.query.WindowKeyQuery;
import org.apache.kafka.streams.query.WindowRangeQuery;
import org.apache.kafka.streams.state.KeyValueIterator;
+import org.apache.kafka.streams.state.ReadOnlyWindowStore;
import org.apache.kafka.streams.state.StateSerdes;
import org.apache.kafka.streams.state.Stores;
import org.apache.kafka.streams.state.TimestampedWindowStoreWithHeaders;
@@ -52,14 +54,19 @@ import java.time.Instant;
import java.util.LinkedList;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicReference;
import static java.time.Duration.ofMillis;
import static org.apache.kafka.common.utils.Utils.mkEntry;
import static org.apache.kafka.common.utils.Utils.mkMap;
import static
org.apache.kafka.streams.state.internals.WindowKeySchema.toStoreKeyBinary;
+import static org.junit.jupiter.api.Assertions.assertArrayEquals;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -253,6 +260,170 @@ public class InMemoryWindowStoreTest extends
AbstractWindowBytesStoreTest {
assertEquals(expected, actual);
}
+ @Test
+ public void readOnlyCommittedShouldHideStagedFetchValue() {
+ final InMemoryWindowStore txnStore = openTransactionalWindowStore();
+ try {
+ final Bytes key = Bytes.wrap("k".getBytes());
+ txnStore.put(key, "v1".getBytes(), 0L);
+ txnStore.commit(java.util.Map.of());
+
+ txnStore.put(key, "v2".getBytes(), WINDOW_SIZE);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> uncommitted =
txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyWindowStore<Bytes, byte[]> committed =
txnStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ assertArrayEquals("v1".getBytes(), uncommitted.fetch(key, 0L));
+ assertArrayEquals("v2".getBytes(), uncommitted.fetch(key,
WINDOW_SIZE));
+ assertArrayEquals("v1".getBytes(), committed.fetch(key, 0L));
+ assertNull(committed.fetch(key, WINDOW_SIZE));
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyFetchRangeShouldRespectIsolationLevel() {
+ final InMemoryWindowStore txnStore = openTransactionalWindowStore();
+ try {
+ final Bytes key = Bytes.wrap("k".getBytes());
+ txnStore.put(key, "v1".getBytes(), 0L);
+ txnStore.commit(java.util.Map.of());
+ txnStore.put(key, "v2".getBytes(), WINDOW_SIZE);
+
+ final ReadOnlyWindowStore<Bytes, byte[]> uncommitted =
txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyWindowStore<Bytes, byte[]> committed =
txnStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ try (WindowStoreIterator<byte[]> it = uncommitted.fetch(key,
Instant.ofEpochMilli(0), Instant.ofEpochMilli(WINDOW_SIZE))) {
+ final List<String> values = new LinkedList<>();
+ while (it.hasNext()) {
+ values.add(new String(it.next().value));
+ }
+ assertEquals(List.of("v1", "v2"), values);
+ }
+ try (WindowStoreIterator<byte[]> it = committed.fetch(key,
Instant.ofEpochMilli(0), Instant.ofEpochMilli(WINDOW_SIZE))) {
+ final List<String> values = new LinkedList<>();
+ while (it.hasNext()) {
+ values.add(new String(it.next().value));
+ }
+ assertEquals(List.of("v1"), values);
+ }
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyAllShouldRespectIsolationLevel() {
+ final InMemoryWindowStore txnStore = openTransactionalWindowStore();
+ try {
+ txnStore.put(Bytes.wrap("k1".getBytes()), "v1".getBytes(), 0L);
+ txnStore.commit(java.util.Map.of());
+ txnStore.put(Bytes.wrap("k2".getBytes()), "v2".getBytes(),
WINDOW_SIZE);
+
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+ txnStore.readOnly(IsolationLevel.READ_UNCOMMITTED).all())
{
+ int count = 0;
+ while (it.hasNext()) {
+ it.next();
+ count++;
+ }
+ assertEquals(2, count);
+ }
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+ txnStore.readOnly(IsolationLevel.READ_COMMITTED).all()) {
+ int count = 0;
+ while (it.hasNext()) {
+ it.next();
+ count++;
+ }
+ assertEquals(1, count);
+ }
+ } finally {
+ txnStore.close();
+ }
+ }
+
+ @Test
+ public void readOnlyCommittedFetchAllShouldNotRaceConcurrentCommit()
throws InterruptedException {
+ // A READ_COMMITTED scan from a non-owner (IQ) thread must observe a
point-in-time snapshot of
+ // the committed segments, taken under the buffer's snapshot
read-lock. Without it the view
+ // iterated the live segment map while commit()'s write-lock-guarded
flushToBase() mutated it,
+ // exposing a half-applied commit. Each commit flips every key to a
single generation, so a
+ // correct snapshot scan must see one uniform generation across the
full key set.
+ final InMemoryWindowStore txnStore = openTransactionalWindowStore();
+ final int numKeys = 300;
+ final int rounds = 300;
+ final AtomicReference<Throwable> readerFailure = new
AtomicReference<>();
+ final AtomicBoolean stop = new AtomicBoolean(false);
+
+ try {
+ for (int i = 0; i < numKeys; i++) {
+ txnStore.put(Bytes.wrap(("k" + i).getBytes()),
"gen0".getBytes(), 0L);
+ }
+ txnStore.commit(java.util.Map.of());
+
+ final Thread reader = new Thread(() -> {
+ final ReadOnlyWindowStore<Bytes, byte[]> committed =
txnStore.readOnly(IsolationLevel.READ_COMMITTED);
+ try {
+ while (!stop.get()) {
+ try (KeyValueIterator<Windowed<Bytes>, byte[]> it =
+ committed.fetchAll(Instant.ofEpochMilli(0),
Instant.ofEpochMilli(RETENTION_PERIOD))) {
+ String generation = null;
+ int count = 0;
+ while (it.hasNext()) {
+ final String value = new
String(it.next().value);
+ if (generation == null) {
+ generation = value;
+ } else {
+ assertEquals(generation, value, "scan
observed a partial commit");
+ }
+ count++;
+ }
+ assertEquals(numKeys, count, "scan observed a
partial commit");
+ }
+ }
+ } catch (final Throwable t) {
+ readerFailure.set(t);
+ }
+ }, "iq-reader");
+ reader.start();
+
+ for (int round = 1; round <= rounds && readerFailure.get() ==
null; round++) {
+ final String generation = "gen" + round;
+ for (int i = 0; i < numKeys; i++) {
+ txnStore.put(Bytes.wrap(("k" + i).getBytes()),
generation.getBytes(), 0L);
+ }
+ txnStore.commit(java.util.Map.of());
+ }
+ stop.set(true);
+ reader.join(TimeUnit.SECONDS.toMillis(30));
+
+ if (readerFailure.get() != null) {
+ throw new AssertionError("READ_COMMITTED fetchAll raced a
concurrent commit", readerFailure.get());
+ }
+ } finally {
+ stop.set(true);
+ txnStore.close();
+ }
+ }
+
+ private InMemoryWindowStore openTransactionalWindowStore() {
+ final Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ final InternalMockProcessorContext<Bytes, byte[]> ctx = new
InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ new Serdes.BytesSerde(),
+ new Serdes.ByteArraySerde(),
+ new StreamsConfig(props)
+ );
+ final InMemoryWindowStore store = new InMemoryWindowStore(
+ "txn-in-memory-window-store", RETENTION_PERIOD, WINDOW_SIZE,
false, "scope");
+ store.init(ctx, store);
+ return store;
+ }
+
@Nested
class InMemoryStoreIQv2Tests {
private static final long WINDOW_SIZE = 10_000L;
