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 8cfdfaaacc9 KAFKA-20500: Add isolation-level reads to versioned stores
(#22682)
8cfdfaaacc9 is described below
commit 8cfdfaaacc90bd7c62c38253931b9c3f7d3593b5
Author: Nick Telford <[email protected]>
AuthorDate: Tue Jun 30 00:16:03 2026 +0100
KAFKA-20500: Add isolation-level reads to versioned stores (#22682)
Part of the KIP-892 interactive-query isolation-level series. Versioned
stores are queryable through IQv2 (`VersionedKeyQuery`,
`MultiVersionedKeyQuery`) and IQv1, but had no way to honour the
configured isolation level. When the underlying `RocksDBStore` is
transactional its accessor consults the staged-write buffer, so a
`READ_COMMITTED` query would incorrectly observe writes still in the
current transaction.
This extends the `readOnly(IsolationLevel)` hook the other store
families already have to versioned stores. Because versioned stores have
no dedicated `ReadOnly*` parent interface, the default is added directly
on `VersionedKeyValueStore`, and `VersionedBytesStore.readOnly` is
covariantly narrowed so wrapper layers keep the versioned read methods.
Reads in `RocksDBVersionedStore` — single-key latest, point-in-time, and
timestamp-range — flow through `LogicalKeyValueSegment` views bound to a
specific `DBAccessor`, so `READ_COMMITTED` bypasses the transaction
buffer via the direct accessor. The metered and change-logging versioned
wrappers gain matching overrides, and `StoreQueryUtils` dispatches
versioned key queries through `readOnly(isolationLevel)`.
Semantic tests assert that `READ_COMMITTED` hides staged writes while
`READ_UNCOMMITTED` exposes them across the single-key, point-in-time,
and timestamp-range read paths.
This branched off the now-merged RocksDB isolation-level read work
(KAFKA-20498) and now applies directly to trunk.
Reviewers: Bill Bejeck <[email protected]>
---
.../kafka/streams/state/VersionedBytesStore.java | 11 ++
.../streams/state/VersionedKeyValueStore.java | 13 +++
.../ChangeLoggingVersionedKeyValueBytesStore.java | 6 +
.../state/internals/LogicalKeyValueSegment.java | 44 +++++++-
.../internals/MeteredVersionedKeyValueStore.java | 90 +++++++++++++++
.../streams/state/internals/RocksDBStore.java | 41 +++++--
.../state/internals/RocksDBVersionedStore.java | 124 +++++++++++++++++++--
.../streams/state/internals/StoreQueryUtils.java | 5 +-
.../VersionedKeyValueToBytesStoreAdapter.java | 6 +
.../state/internals/RocksDBVersionedStoreTest.java | 84 ++++++++++++++
10 files changed, 401 insertions(+), 23 deletions(-)
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/VersionedBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/VersionedBytesStore.java
index e840824577e..b1d7540437d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/VersionedBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/VersionedBytesStore.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.utils.Bytes;
/**
@@ -38,4 +39,14 @@ public interface VersionedBytesStore extends
KeyValueStore<Bytes, byte[]>, Times
* The analog of {@link VersionedKeyValueStore#delete(Object, long)}.
*/
byte[] delete(Bytes key, long timestamp);
+
+ /**
+ * Return a read-only view of this store bound to the given {@link
IsolationLevel}.
+ * Covariantly narrows {@link KeyValueStore#readOnly(IsolationLevel)} so
callers retain
+ * access to the versioned read methods ({@link #get(Bytes, long)} and
friends).
+ */
+ @Override
+ default VersionedBytesStore readOnly(final IsolationLevel isolationLevel) {
+ return this;
+ }
}
\ No newline at end of file
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java
index 40faaf003d3..ebd4fc7e34d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/VersionedKeyValueStore.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.streams.errors.InvalidStateStoreException;
import org.apache.kafka.streams.processor.StateStore;
@@ -134,4 +135,16 @@ public interface VersionedKeyValueStore<K, V> extends
StateStore {
* @throws InvalidStateStoreException if the store is not initialized
*/
VersionedRecord<V> get(K key, long asOfTimestamp);
+
+ /**
+ * Return a read-only view of this store bound to the given {@link
IsolationLevel}.
+ * See {@link ReadOnlyKeyValueStore#readOnly(IsolationLevel)} for
semantics.
+ * <p>
+ * Unlike the other store families, versioned stores have no dedicated
{@code ReadOnly*}
+ * parent interface — the view is just another {@code
VersionedKeyValueStore} whose mutating
+ * operations should not be invoked from interactive-query threads.
+ */
+ default VersionedKeyValueStore<K, V> readOnly(final IsolationLevel
isolationLevel) {
+ return this;
+ }
}
\ No newline at end of file
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingVersionedKeyValueBytesStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingVersionedKeyValueBytesStore.java
index bd35648210d..a391dcee8c0 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingVersionedKeyValueBytesStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/ChangeLoggingVersionedKeyValueBytesStore.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.utils.Bytes;
@@ -57,6 +58,11 @@ public class ChangeLoggingVersionedKeyValueBytesStore
extends ChangeLoggingKeyVa
return oldValue;
}
+ @Override
+ public VersionedBytesStore readOnly(final IsolationLevel isolationLevel) {
+ return inner.readOnly(isolationLevel);
+ }
+
@Override public void log(final Bytes key, final byte[] value, final long
timestamp, final Headers headers) {
internalContext.logChange(
name(),
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java
index ba6215ecca0..2aa5182fc2d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.BytesSerializer;
import org.apache.kafka.common.utils.Bytes;
@@ -60,19 +61,45 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
private final String name;
private final RocksDBStore physicalStore;
private final PrefixKeyFormatter prefixKeyFormatter;
+ // Non-null for read-only views produced by {@link
#readOnly(IsolationLevel)}: all reads go
+ // through this accessor (bypassing any transaction buffer for
READ_COMMITTED); writes are
+ // disallowed. Null for regular segments, which use the physicalStore's
current accessor.
+ private final RocksDBStore.DBAccessor readAccessor;
final Set<KeyValueIterator<Bytes, byte[]>> openIterators =
Collections.synchronizedSet(new HashSet<>());
LogicalKeyValueSegment(final long id,
final String name,
final RocksDBStore physicalStore) {
+ this(id, name, physicalStore, null);
+ }
+
+ private LogicalKeyValueSegment(final long id,
+ final String name,
+ final RocksDBStore physicalStore,
+ final RocksDBStore.DBAccessor readAccessor)
{
this.id = id;
this.name = name;
this.physicalStore = Objects.requireNonNull(physicalStore);
-
+ this.readAccessor = readAccessor;
this.prefixKeyFormatter = new
PrefixKeyFormatter(serializeLongToBytes(id));
}
+ /**
+ * Returns a read-only view of this segment bound to the given isolation
level. Reads go
+ * through the accessor appropriate for {@code level}; mutating calls
throw.
+ */
+ @Override
+ public LogicalKeyValueSegment readOnly(final IsolationLevel level) {
+ return new LogicalKeyValueSegment(id, name, physicalStore,
physicalStore.dbAccessor.readOnly(level));
+ }
+
+ private void rejectIfReadOnly() {
+ if (readAccessor != null) {
+ throw new UnsupportedOperationException("Write operations are not
supported on a read-only segment view");
+ }
+ }
+
@Override
public long id() {
return id;
@@ -80,6 +107,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized void destroy() {
+ rejectIfReadOnly();
if (id < 0) {
throw new IllegalStateException("Negative segment ID indicates a
reserved segment, "
+ "which should not be destroyed. Reserved segments are
cleaned up only when "
@@ -95,6 +123,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized void deleteRange(final Bytes keyFrom, final Bytes
keyTo) {
+ rejectIfReadOnly();
physicalStore.deleteRange(
prefixKeyFormatter.addPrefix(keyFrom),
prefixKeyFormatter.addPrefix(keyTo));
@@ -102,6 +131,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized void put(final Bytes key, final byte[] value) {
+ rejectIfReadOnly();
physicalStore.put(
prefixKeyFormatter.addPrefix(key),
value);
@@ -109,6 +139,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized byte[] putIfAbsent(final Bytes key, final byte[]
value) {
+ rejectIfReadOnly();
return physicalStore.putIfAbsent(
prefixKeyFormatter.addPrefix(key),
value);
@@ -116,6 +147,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized void putAll(final List<KeyValue<Bytes, byte[]>>
entries) {
+ rejectIfReadOnly();
physicalStore.putAll(entries.stream()
.map(kv -> new KeyValue<>(
prefixKeyFormatter.addPrefix(kv.key),
@@ -125,6 +157,7 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
@Override
public synchronized byte[] delete(final Bytes key) {
+ rejectIfReadOnly();
return physicalStore.delete(prefixKeyFormatter.addPrefix(key));
}
@@ -184,13 +217,18 @@ public class LogicalKeyValueSegment implements Segment,
VersionedStoreSegment {
}
private synchronized byte[] get(final Bytes key, final Optional<Snapshot>
snapshot) {
+ final Bytes prefixed = prefixKeyFormatter.addPrefix(key);
if (snapshot.isPresent()) {
try (ReadOptions readOptions = new ReadOptions()) {
readOptions.setSnapshot(snapshot.get());
- return physicalStore.get(prefixKeyFormatter.addPrefix(key),
readOptions);
+ return readAccessor == null
+ ? physicalStore.get(prefixed, readOptions)
+ : physicalStore.get(prefixed, readOptions, readAccessor);
}
} else {
- return physicalStore.get(prefixKeyFormatter.addPrefix(key));
+ return readAccessor == null
+ ? physicalStore.get(prefixed)
+ : physicalStore.get(prefixed, readAccessor);
}
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java
index 9cd8b9e1861..2f6f5380c1a 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/MeteredVersionedKeyValueStore.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.utils.Bytes;
@@ -403,4 +404,93 @@ public class MeteredVersionedKeyValueStore<K, V>
public Position getPosition() {
return internal.getPosition();
}
+
+ @Override
+ public VersionedKeyValueStore<K, V> readOnly(final IsolationLevel
isolationLevel) {
+ return new ReadOnlyView(wrapped().readOnly(isolationLevel));
+ }
+
+ /**
+ * Read-only view that re-applies this store's serdes and get sensor on
top of a configurable
+ * underlying {@link VersionedBytesStore}. Used so that IQ reads at a
given isolation level
+ * preserve the Metered layer's metrics and (de)serialisation behaviour.
Mutating operations
+ * throw; lifecycle methods that only report state delegate to the
enclosing store.
+ */
+ private final class ReadOnlyView implements VersionedKeyValueStore<K, V> {
+
+ private final VersionedBytesStore underlying;
+
+ ReadOnlyView(final VersionedBytesStore underlying) {
+ this.underlying = underlying;
+ }
+
+ @Override
+ public VersionedRecord<V> get(final K key) {
+ Objects.requireNonNull(key, "key cannot be null");
+ try {
+ final ValueAndTimestamp<V> valueAndTimestamp =
maybeMeasureLatency(
+ () ->
internal.deserializeValue(underlying.get(internal.serializeKey(key))),
+ internal.time,
+ internal.getSensor
+ );
+ return valueAndTimestamp == null
+ ? null
+ : new VersionedRecord<>(valueAndTimestamp.value(),
valueAndTimestamp.timestamp());
+ } catch (final ProcessorStateException e) {
+ throw new
ProcessorStateException(String.format(e.getMessage(), key), e);
+ }
+ }
+
+ @Override
+ public VersionedRecord<V> get(final K key, final long asOfTimestamp) {
+ Objects.requireNonNull(key, "key cannot be null");
+ try {
+ final ValueAndTimestamp<V> valueAndTimestamp =
maybeMeasureLatency(
+ () ->
internal.deserializeValue(underlying.get(internal.serializeKey(key),
asOfTimestamp)),
+ internal.time,
+ internal.getSensor
+ );
+ return valueAndTimestamp == null
+ ? null
+ : new VersionedRecord<>(valueAndTimestamp.value(),
valueAndTimestamp.timestamp());
+ } catch (final ProcessorStateException e) {
+ throw new
ProcessorStateException(String.format(e.getMessage(), key), e);
+ }
+ }
+
+ @Override
+ public long put(final K key, final V value, final long timestamp) {
+ throw new UnsupportedOperationException("put is not supported on a
read-only view");
+ }
+
+ @Override
+ public VersionedRecord<V> delete(final K key, final long timestamp) {
+ throw new UnsupportedOperationException("delete is not supported
on a read-only view");
+ }
+
+ @Override
+ public String name() {
+ return MeteredVersionedKeyValueStore.this.name();
+ }
+
+ @Override
+ public void init(final StateStoreContext stateStoreContext, final
StateStore root) {
+ throw new UnsupportedOperationException("init is not supported on
a read-only view");
+ }
+
+ @Override
+ public void close() {
+ throw new UnsupportedOperationException("close is not supported on
a read-only view");
+ }
+
+ @Override
+ public boolean persistent() {
+ return MeteredVersionedKeyValueStore.this.persistent();
+ }
+
+ @Override
+ public boolean isOpen() {
+ return MeteredVersionedKeyValueStore.this.isOpen();
+ }
+ }
}
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
index 8b16da8003c..417c36e8547 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java
@@ -759,14 +759,26 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
@Override
public ReadOnlyKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
- Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
- final DBAccessor viewAccessor;
- if (isolationLevel == IsolationLevel.READ_COMMITTED && dbAccessor
instanceof TransactionalDBAccessor) {
- viewAccessor = ((TransactionalDBAccessor) dbAccessor).underlying;
- } else {
- viewAccessor = dbAccessor;
+ return new ReadOnlyView(dbAccessor.readOnly(isolationLevel));
+ }
+
+ // Read helpers for isolation-level views that sit above this store (e.g.
LogicalKeyValueSegment.readOnly).
+ byte[] get(final Bytes key, final DBAccessor accessor) {
+ validateStoreOpen();
+ try {
+ return cfAccessor.get(accessor, key.get());
+ } catch (final RocksDBException e) {
+ throw new ProcessorStateException("Error while getting value for
key from store " + name, e);
+ }
+ }
+
+ byte[] get(final Bytes key, final ReadOptions readOptions, final
DBAccessor accessor) {
+ validateStoreOpen();
+ try {
+ return cfAccessor.get(accessor, key.get(), readOptions);
+ } catch (final RocksDBException e) {
+ throw new ProcessorStateException("Error while getting value for
key from store " + name, e);
}
- return new ReadOnlyView(viewAccessor);
}
/**
@@ -1065,6 +1077,11 @@ public class RocksDBStore implements
KeyValueStore<Bytes, byte[]>, BatchWritingS
void reset();
void close();
+ default DBAccessor readOnly(final IsolationLevel isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ return this;
+ }
+
default ManagedKeyValueIterator<Bytes, byte[]> all(final
ColumnFamilyHandle cf, final String storeName, final boolean forward) {
final RocksIterator iter = newIterator(cf);
if (forward) {
@@ -1258,6 +1275,16 @@ public class RocksDBStore implements
KeyValueStore<Bytes, byte[]>, BatchWritingS
underlying.close();
}
+ @Override
+ public DBAccessor readOnly(final IsolationLevel isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ if (isolationLevel == IsolationLevel.READ_COMMITTED) {
+ return underlying;
+ } else {
+ return this;
+ }
+ }
+
@Override
public ManagedKeyValueIterator<Bytes, byte[]> all(final
ColumnFamilyHandle cf, final String storeName, final boolean forward) {
return buffer.all(cf, forward);
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
index c4c26d047f5..6477427e66f 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.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;
@@ -183,11 +184,15 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
@Override
public VersionedRecord<byte[]> get(final Bytes key) {
+ return get(key, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private VersionedRecord<byte[]> get(final Bytes key, final IsolationLevel
level) {
Objects.requireNonNull(key, "key cannot be null");
validateStoreOpen();
// latest value (if present) is guaranteed to be in the latest value
store
- final byte[] rawLatestValueAndTimestamp = latestValueStore.get(key);
+ final byte[] rawLatestValueAndTimestamp =
latestValueStore(level).get(key);
if (rawLatestValueAndTimestamp != null) {
return new VersionedRecord<>(
LatestValueFormatter.value(rawLatestValueAndTimestamp),
@@ -200,14 +205,20 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
@Override
public VersionedRecord<byte[]> get(final Bytes key, final long
asOfTimestamp) {
+ return get(key, asOfTimestamp, IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ private VersionedRecord<byte[]> get(final Bytes key, final long
asOfTimestamp, final IsolationLevel level) {
Objects.requireNonNull(key, "key cannot be null");
validateStoreOpen();
+ final LogicalKeyValueSegment latestView = latestValueStore(level);
+
if (asOfTimestamp < observedStreamTime - historyRetention) {
// history retention exceeded. we still check the latest value
store in case the
// latest record version satisfies the timestamp bound, in which
case it should
// still be returned (i.e., the latest record version per key
never expires).
- final byte[] rawLatestValueAndTimestamp =
latestValueStore.get(key);
+ final byte[] rawLatestValueAndTimestamp = latestView.get(key);
if (rawLatestValueAndTimestamp != null) {
final long latestTimestamp =
LatestValueFormatter.timestamp(rawLatestValueAndTimestamp);
if (latestTimestamp <= asOfTimestamp) {
@@ -227,7 +238,7 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
}
// first check the latest value store
- final byte[] rawLatestValueAndTimestamp = latestValueStore.get(key);
+ final byte[] rawLatestValueAndTimestamp = latestView.get(key);
if (rawLatestValueAndTimestamp != null) {
final long latestTimestamp =
LatestValueFormatter.timestamp(rawLatestValueAndTimestamp);
if (latestTimestamp <= asOfTimestamp) {
@@ -236,8 +247,7 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
}
// check segment stores
- final List<LogicalKeyValueSegment> segments =
segmentStores.segments(asOfTimestamp, Long.MAX_VALUE, false);
- for (final LogicalKeyValueSegment segment : segments) {
+ for (final LogicalKeyValueSegment segment :
viewSegments(segmentStores.segments(asOfTimestamp, Long.MAX_VALUE, false),
level)) {
final byte[] rawSegmentValue = segment.get(key);
if (rawSegmentValue != null) {
final long nextTs =
RocksDBVersionedStoreSegmentValueFormatter.nextTimestamp(rawSegmentValue);
@@ -272,31 +282,123 @@ public class RocksDBVersionedStore implements
VersionedKeyValueStore<Bytes, byte
return null;
}
- @SuppressWarnings("unchecked")
VersionedRecordIterator<byte[]> get(final Bytes key, final long
fromTimestamp, final long toTimestamp, final ResultOrder order) {
+ return get(key, fromTimestamp, toTimestamp, order,
IsolationLevel.READ_UNCOMMITTED);
+ }
+
+ VersionedRecordIterator<byte[]> get(final Bytes key, final long
fromTimestamp, final long toTimestamp,
+ final ResultOrder order, final
IsolationLevel level) {
validateStoreOpen();
+ final LogicalKeyValueSegment latestView = latestValueStore(level);
+
if (toTimestamp < observedStreamTime - historyRetention) {
// history retention exceeded. we still check the latest value
store in case the
// latest record version satisfies the timestamp bound, in which
case it should
// still be returned (i.e., the latest record version per key
never expires).
- return new
LogicalSegmentIterator(Collections.singletonList(latestValueStore).listIterator(),
key, fromTimestamp, toTimestamp, order);
+ return new
LogicalSegmentIterator(Collections.singletonList(latestView).listIterator(),
key, fromTimestamp, toTimestamp, order);
} else {
final List<LogicalKeyValueSegment> segments = new ArrayList<>();
// add segment stores
// consider the search lower bound as -INF (LONG.MIN_VALUE) to
find the record that has been inserted before the {@code fromTimestamp}
// but is still valid in query specified time interval.
if (order.equals(ResultOrder.ASCENDING)) {
- segments.addAll(segmentStores.segments(Long.MIN_VALUE,
toTimestamp, true));
- segments.add(latestValueStore);
+
segments.addAll(viewSegments(segmentStores.segments(Long.MIN_VALUE,
toTimestamp, true), level));
+ segments.add(latestView);
} else {
- segments.add(latestValueStore);
- segments.addAll(segmentStores.segments(Long.MIN_VALUE,
toTimestamp, false));
+ segments.add(latestView);
+
segments.addAll(viewSegments(segmentStores.segments(Long.MIN_VALUE,
toTimestamp, false), level));
}
return new LogicalSegmentIterator(segments.listIterator(), key,
fromTimestamp, toTimestamp, order);
}
}
+ private LogicalKeyValueSegment latestValueStore(final IsolationLevel
level) {
+ return level == IsolationLevel.READ_UNCOMMITTED ? latestValueStore :
latestValueStore.readOnly(level);
+ }
+
+ private static List<LogicalKeyValueSegment> viewSegments(final
List<LogicalKeyValueSegment> segments,
+ final
IsolationLevel level) {
+ if (level == IsolationLevel.READ_UNCOMMITTED) {
+ return segments;
+ }
+ final List<LogicalKeyValueSegment> views = new
ArrayList<>(segments.size());
+ for (final LogicalKeyValueSegment segment : segments) {
+ views.add(segment.readOnly(level));
+ }
+ return views;
+ }
+
+ @Override
+ public VersionedKeyValueStore<Bytes, byte[]> readOnly(final IsolationLevel
isolationLevel) {
+ Objects.requireNonNull(isolationLevel, "isolationLevel cannot be
null");
+ if (isolationLevel == IsolationLevel.READ_UNCOMMITTED) {
+ return this;
+ }
+ return new ReadOnlyView(isolationLevel);
+ }
+
+ /**
+ * Read-only view of this store bound to an isolation level. Read methods
delegate to
+ * the private {@code get(...)} helpers passing the level; write methods
and other
+ * lifecycle operations throw or are no-ops, since the view is only used
by IQ.
+ */
+ private final class ReadOnlyView implements VersionedKeyValueStore<Bytes,
byte[]> {
+
+ private final IsolationLevel level;
+
+ ReadOnlyView(final IsolationLevel level) {
+ this.level = level;
+ }
+
+ @Override
+ public VersionedRecord<byte[]> get(final Bytes key) {
+ return RocksDBVersionedStore.this.get(key, level);
+ }
+
+ @Override
+ public VersionedRecord<byte[]> get(final Bytes key, final long
asOfTimestamp) {
+ return RocksDBVersionedStore.this.get(key, asOfTimestamp, level);
+ }
+
+ @Override
+ public long put(final Bytes key, final byte[] value, final long
timestamp) {
+ throw new UnsupportedOperationException("put not supported on a
read-only view");
+ }
+
+ @Override
+ public VersionedRecord<byte[]> delete(final Bytes key, final long
timestamp) {
+ throw new UnsupportedOperationException("delete not supported on a
read-only view");
+ }
+
+ @Override
+ public String name() {
+ return RocksDBVersionedStore.this.name();
+ }
+
+ @Override
+ public void init(final StateStoreContext stateStoreContext, final
StateStore root) {
+ throw new UnsupportedOperationException("init not supported on a
read-only view");
+ }
+
+ @SuppressWarnings("deprecation")
+ @Override
+ public void flush() { }
+
+ @Override
+ public void close() { }
+
+ @Override
+ public boolean persistent() {
+ return RocksDBVersionedStore.this.persistent();
+ }
+
+ @Override
+ public boolean isOpen() {
+ return RocksDBVersionedStore.this.isOpen();
+ }
+ }
+
@Override
public String name() {
return name;
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
index ab23baa765b..c713c546d70 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/StoreQueryUtils.java
@@ -373,7 +373,7 @@ public final class StoreQueryUtils {
) {
if (store instanceof VersionedKeyValueStore) {
final VersionedKeyValueStore<Bytes, byte[]> versionedKeyValueStore
=
- (VersionedKeyValueStore<Bytes, byte[]>) store;
+ ((VersionedKeyValueStore<Bytes, byte[]>)
store).readOnly(config.getIsolationLevel());
final VersionedKeyQuery<Bytes, byte[]> rawKeyQuery =
(VersionedKeyQuery<Bytes, byte[]>) query;
try {
@@ -413,7 +413,8 @@ public final class StoreQueryUtils {
rawKeyQuery.key(),
rawKeyQuery.fromTime().get().toEpochMilli(),
rawKeyQuery.toTime().get().toEpochMilli(),
- rawKeyQuery.resultOrder()
+ rawKeyQuery.resultOrder(),
+ config.getIsolationLevel()
);
return (QueryResult<R>) QueryResult.forResult(segmentIterator);
} catch (final Exception e) {
diff --git
a/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
index ee422bd837e..9b149fa904d 100644
---
a/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
+++
b/streams/src/main/java/org/apache/kafka/streams/state/internals/VersionedKeyValueToBytesStoreAdapter.java
@@ -16,6 +16,7 @@
*/
package org.apache.kafka.streams.state.internals;
+import org.apache.kafka.common.IsolationLevel;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.Serde;
import org.apache.kafka.common.serialization.Serdes.ByteArraySerde;
@@ -84,6 +85,11 @@ public class VersionedKeyValueToBytesStoreAdapter implements
VersionedBytesStore
return serializeAsBytes(versionedRecord);
}
+ @Override
+ public VersionedBytesStore readOnly(final IsolationLevel isolationLevel) {
+ return new
VersionedKeyValueToBytesStoreAdapter(inner.readOnly(isolationLevel));
+ }
+
@Override
public String name() {
return inner.name();
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java
index 340ca7eb216..ad74da70d49 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStoreTest.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.internals.RecordHeaders;
@@ -30,6 +31,7 @@ import org.apache.kafka.common.utils.Bytes;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.query.Position;
import org.apache.kafka.streams.query.ResultOrder;
+import org.apache.kafka.streams.state.VersionedKeyValueStore;
import org.apache.kafka.streams.state.VersionedRecord;
import org.apache.kafka.streams.state.VersionedRecordIterator;
import org.apache.kafka.test.InternalMockProcessorContext;
@@ -833,6 +835,88 @@ public class RocksDBVersionedStoreTest {
verifyExpiredRecordSensor(1);
}
+ @Test
+ public void readOnlyCommittedShouldHideStagedLatestPut() {
+ reopenWithTransactionalEOS();
+
+ putToStore("k", "v1", BASE_TIMESTAMP,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+ store.commit(Map.of());
+ putToStore("k", "v2", BASE_TIMESTAMP + 1,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+
+ final VersionedKeyValueStore<Bytes, byte[]> uncommitted =
store.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final VersionedKeyValueStore<Bytes, byte[]> committed =
store.readOnly(IsolationLevel.READ_COMMITTED);
+
+ final Bytes key = new Bytes(STRING_SERIALIZER.serialize(null, "k"));
+ final VersionedRecord<byte[]> uLatest = uncommitted.get(key);
+ assertThat(STRING_DESERIALIZER.deserialize(null, uLatest.value()),
equalTo("v2"));
+ assertThat(uLatest.timestamp(), equalTo(BASE_TIMESTAMP + 1));
+
+ final VersionedRecord<byte[]> cLatest = committed.get(key);
+ assertThat(STRING_DESERIALIZER.deserialize(null, cLatest.value()),
equalTo("v1"));
+ assertThat(cLatest.timestamp(), equalTo(BASE_TIMESTAMP));
+ }
+
+ @Test
+ public void readOnlyCommittedShouldHideStagedTimestampedPut() {
+ reopenWithTransactionalEOS();
+
+ putToStore("k", "v1", BASE_TIMESTAMP,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+ store.commit(Map.of());
+ putToStore("k", "v2", BASE_TIMESTAMP + 2,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+
+ final Bytes key = new Bytes(STRING_SERIALIZER.serialize(null, "k"));
+
+ final VersionedRecord<byte[]> uAtNew =
store.readOnly(IsolationLevel.READ_UNCOMMITTED).get(key, BASE_TIMESTAMP + 2);
+ assertThat(STRING_DESERIALIZER.deserialize(null, uAtNew.value()),
equalTo("v2"));
+
+ final VersionedRecord<byte[]> cAtNew =
store.readOnly(IsolationLevel.READ_COMMITTED).get(key, BASE_TIMESTAMP + 2);
+ assertThat(STRING_DESERIALIZER.deserialize(null, cAtNew.value()),
equalTo("v1"));
+ }
+
+ @Test
+ public void packagePrivateTimestampRangeGetShouldRespectIsolationLevel() {
+ reopenWithTransactionalEOS();
+
+ putToStore("k", "v1", BASE_TIMESTAMP,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+ store.commit(Map.of());
+ putToStore("k", "v2", BASE_TIMESTAMP + 2,
PUT_RETURN_CODE_VALID_TO_UNDEFINED);
+
+ final Bytes key = new Bytes(STRING_SERIALIZER.serialize(null, "k"));
+
+ final List<String> uncommitted = new ArrayList<>();
+ try (VersionedRecordIterator<byte[]> it = store.get(
+ key, BASE_TIMESTAMP, BASE_TIMESTAMP + 2,
ResultOrder.ASCENDING, IsolationLevel.READ_UNCOMMITTED)) {
+ while (it.hasNext()) {
+ uncommitted.add(STRING_DESERIALIZER.deserialize(null,
it.next().value()));
+ }
+ }
+ final List<String> committed = new ArrayList<>();
+ try (VersionedRecordIterator<byte[]> it = store.get(
+ key, BASE_TIMESTAMP, BASE_TIMESTAMP + 2,
ResultOrder.ASCENDING, IsolationLevel.READ_COMMITTED)) {
+ while (it.hasNext()) {
+ committed.add(STRING_DESERIALIZER.deserialize(null,
it.next().value()));
+ }
+ }
+ assertThat(uncommitted, equalTo(List.of("v1", "v2")));
+ assertThat(committed, equalTo(List.of("v1")));
+ }
+
+ private void reopenWithTransactionalEOS() {
+ store.close();
+ final java.util.Properties props = StreamsTestUtils.getStreamsConfig();
+ props.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+ props.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ context = new InternalMockProcessorContext<>(
+ TestUtils.tempDirectory(),
+ Serdes.String(),
+ Serdes.String(),
+ new StreamsConfig(props)
+ );
+ context.setTime(BASE_TIMESTAMP);
+ store = new RocksDBVersionedStore(STORE_NAME, METRICS_SCOPE,
HISTORY_RETENTION, SEGMENT_INTERVAL);
+ store.init(context, store);
+ }
+
@Test
public void shouldMigrateExistingPositionFromFile() {
final Position position = Position.fromMap(mkMap(mkEntry("topic",
mkMap(mkEntry(0, 1L)))));
