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 81de9daf47c KAFKA-20498: Add isolation-level reads to RocksDBStore
(#22653)
81de9daf47c is described below
commit 81de9daf47c7734c77669fd386f2f7a9f65e22c2
Author: Nick Telford <[email protected]>
AuthorDate: Thu Jun 25 19:05:43 2026 +0100
KAFKA-20498: Add isolation-level reads to RocksDBStore (#22653)
Interactive queries against a transactional RocksDB store (EOS) need to
choose whether they observe writes staged in the current, uncommitted
transaction. This adds a `readOnly(IsolationLevel)` override to
`RocksDBStore` that returns a view bound to a specific `DBAccessor`:
`READ_COMMITTED` reads through the underlying `DirectDBAccessor`,
bypassing the transaction buffer, so the query sees only committed data;
`READ_UNCOMMITTED` (and the default for non-transactional stores) uses
the active accessor, preserving existing behaviour.
The returned `ReadOnlyView` shares `cfAccessor` and `openIterators` with
the store, so reads go through the same iterator-tracking path while the
view exposes no writes.
Includes `readOnly(IsolationLevel)` semantic tests covering `get`,
`range`/`all`, and `prefixScan` that stage writes without committing and
assert the `READ_COMMITTED`/`READ_UNCOMMITTED` divergence, reopening the
store under an EOS context with `enable.transactional.statestores=true`
to exercise the transaction buffer.
Reviewers: Bill Bejeck <[email protected]>
---
.../streams/state/internals/RocksDBStore.java | 103 ++++++++++++
.../streams/state/internals/RocksDBStoreTest.java | 185 +++++++++++++++++++++
2 files changed, 288 insertions(+)
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 e2fb110dcd1..7ed94b0f35a 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
@@ -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.RecordingLevel;
import org.apache.kafka.common.serialization.Serializer;
@@ -41,6 +42,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.apache.kafka.streams.state.RocksDBConfigSetter;
import org.apache.kafka.streams.state.internals.metrics.RocksDBMetricsRecorder;
@@ -746,6 +748,107 @@ public class RocksDBStore implements KeyValueStore<Bytes,
byte[]>, BatchWritingS
return 0;
}
+ @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(viewAccessor);
+ }
+
+ /**
+ * Read-only view of this store bound to a specific {@link DBAccessor}.
Reads go through the
+ * chosen accessor so IQ callers can pick {@code READ_COMMITTED} (direct)
or
+ * {@code READ_UNCOMMITTED} (through the transaction buffer) without
affecting the
+ * processor-thread's view of the store.
+ */
+ private final class ReadOnlyView implements ReadOnlyKeyValueStore<Bytes,
byte[]> {
+
+ private final DBAccessor viewAccessor;
+
+ ReadOnlyView(final DBAccessor viewAccessor) {
+ this.viewAccessor = viewAccessor;
+ }
+
+ @Override
+ public byte[] get(final Bytes key) {
+ Objects.requireNonNull(key, "key cannot be null");
+ validateStoreOpen();
+ try {
+ return cfAccessor.get(viewAccessor, key.get());
+ } catch (final RocksDBException e) {
+ throw new ProcessorStateException("Error while getting value
for key from store " + name, e);
+ }
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> range(final Bytes from, final
Bytes to) {
+ return doRange(from, to, true);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseRange(final Bytes from,
final Bytes to) {
+ return doRange(from, to, false);
+ }
+
+ private KeyValueIterator<Bytes, byte[]> doRange(final Bytes from,
final Bytes to, final boolean forward) {
+ if (Objects.nonNull(from) && Objects.nonNull(to) &&
from.compareTo(to) > 0) {
+ return KeyValueIterators.emptyIterator();
+ }
+ validateStoreOpen();
+ final ManagedKeyValueIterator<Bytes, byte[]> iter =
cfAccessor.range(viewAccessor, from, to, forward);
+ openIterators.add(iter);
+ iter.onClose(() -> openIterators.remove(iter));
+ return iter;
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> all() {
+ return doAll(true);
+ }
+
+ @Override
+ public KeyValueIterator<Bytes, byte[]> reverseAll() {
+ return doAll(false);
+ }
+
+ private KeyValueIterator<Bytes, byte[]> doAll(final boolean forward) {
+ validateStoreOpen();
+ final ManagedKeyValueIterator<Bytes, byte[]> iter =
cfAccessor.all(viewAccessor, forward);
+ openIterators.add(iter);
+ iter.onClose(() -> openIterators.remove(iter));
+ return iter;
+ }
+
+ @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");
+ validateStoreOpen();
+ final Bytes prefixBytes =
Bytes.wrap(prefixKeySerializer.serialize(null, prefix));
+ final ManagedKeyValueIterator<Bytes, byte[]> iter =
cfAccessor.prefixScan(viewAccessor, prefixBytes);
+ openIterators.add(iter);
+ iter.onClose(() -> openIterators.remove(iter));
+ return iter;
+ }
+
+ @Override
+ public long approximateNumEntries() {
+ validateStoreOpen();
+ try {
+ final long n = cfAccessor.approximateNumEntries(viewAccessor);
+ return isOverflowing(n) ? Long.MAX_VALUE : n;
+ } catch (final RocksDBException e) {
+ throw new ProcessorStateException("Error fetching property
from store " + name, e);
+ }
+ }
+ }
+
@Override
public long approximateNumEntries() {
validateStoreOpen();
diff --git
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
index 7734646a5c2..454aac15e9b 100644
---
a/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java
+++
b/streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.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.TopicPartition;
@@ -54,6 +55,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.streams.state.RocksDBConfigSetter;
import org.apache.kafka.streams.state.StoreBuilder;
import org.apache.kafka.streams.state.Stores;
@@ -220,6 +222,13 @@ public class RocksDBStoreTest extends
AbstractKeyValueStoreTest {
return getProcessorContext(stateDir, streamsProps);
}
+ private InternalMockProcessorContext<?, ?>
getTransactionalEOSProcessorContext(final File stateDir) {
+ final Properties streamsProps = StreamsTestUtils.getStreamsConfig();
+ streamsProps.setProperty(StreamsConfig.PROCESSING_GUARANTEE_CONFIG,
StreamsConfig.EXACTLY_ONCE_V2);
+
streamsProps.setProperty(StreamsConfig.TRANSACTIONAL_STATE_STORES_CONFIG,
"true");
+ return getProcessorContext(stateDir, streamsProps);
+ }
+
@Test
public void
shouldAddValueProvidersWithoutStatisticsToInjectedMetricsRecorderWhenRecordingLevelInfo()
{
rocksDBStore = getRocksDBStoreWithRocksDBMetricsRecorder();
@@ -1334,6 +1343,182 @@ public class RocksDBStoreTest extends
AbstractKeyValueStoreTest {
}
}
+ @Test
+ public void
readOnlyCommittedShouldHideStagedPutWhileUncommittedExposesIt() {
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ final Bytes key = new Bytes(stringSerializer.serialize(null, "k"));
+ rocksDBStore.put(key, stringSerializer.serialize(null, "committed"));
+ rocksDBStore.commit(Map.of());
+
+ rocksDBStore.put(key, stringSerializer.serialize(null, "staged"));
+
+ final ReadOnlyKeyValueStore<Bytes, byte[]> uncommitted =
rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> committed =
rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ assertEquals("staged", stringDeserializer.deserialize(null,
uncommitted.get(key)));
+ assertEquals("committed", stringDeserializer.deserialize(null,
committed.get(key)));
+ }
+
+ @Test
+ public void readOnlyCommittedShouldNotSeeStagedDelete() {
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ final Bytes key = new Bytes(stringSerializer.serialize(null, "k"));
+ rocksDBStore.put(key, stringSerializer.serialize(null, "v"));
+ rocksDBStore.commit(Map.of());
+
+ rocksDBStore.delete(key);
+
+
assertNull(rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(key));
+ assertEquals("v", stringDeserializer.deserialize(null,
+ rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED).get(key)));
+ }
+
+ @Test
+ public void readOnlyRangeAndAllShouldRespectIsolationLevel() {
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ final Bytes k1 = new Bytes(stringSerializer.serialize(null, "k1"));
+ final Bytes k2 = new Bytes(stringSerializer.serialize(null, "k2"));
+ final Bytes k3 = new Bytes(stringSerializer.serialize(null, "k3"));
+ rocksDBStore.put(k1, stringSerializer.serialize(null, "a"));
+ rocksDBStore.put(k2, stringSerializer.serialize(null, "b"));
+ rocksDBStore.commit(Map.of());
+
+ rocksDBStore.put(k3, stringSerializer.serialize(null, "c"));
+ rocksDBStore.put(k1, stringSerializer.serialize(null, "a2"));
+
+ final ReadOnlyKeyValueStore<Bytes, byte[]> uncommitted =
rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> committed =
rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ final List<KeyValue<String, String>> uncommittedAll;
+ try (KeyValueIterator<Bytes, byte[]> it = uncommitted.all()) {
+ uncommittedAll = getDeserializedList(it);
+ }
+ final List<KeyValue<String, String>> committedAll;
+ try (KeyValueIterator<Bytes, byte[]> it = committed.all()) {
+ committedAll = getDeserializedList(it);
+ }
+ assertEquals(List.of(KeyValue.pair("k1", "a2"), KeyValue.pair("k2",
"b"), KeyValue.pair("k3", "c")), uncommittedAll);
+ assertEquals(List.of(KeyValue.pair("k1", "a"), KeyValue.pair("k2",
"b")), committedAll);
+
+ final List<KeyValue<String, String>> uncommittedRange;
+ try (KeyValueIterator<Bytes, byte[]> it = uncommitted.range(k1, k3)) {
+ uncommittedRange = getDeserializedList(it);
+ }
+ final List<KeyValue<String, String>> committedRange;
+ try (KeyValueIterator<Bytes, byte[]> it = committed.range(k1, k3)) {
+ committedRange = getDeserializedList(it);
+ }
+ assertEquals(List.of(KeyValue.pair("k1", "a2"), KeyValue.pair("k2",
"b"), KeyValue.pair("k3", "c")), uncommittedRange);
+ assertEquals(List.of(KeyValue.pair("k1", "a"), KeyValue.pair("k2",
"b")), committedRange);
+ }
+
+ @Test
+ public void readOnlyReverseRangeAndReverseAllShouldRespectIsolationLevel()
{
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ final Bytes k1 = new Bytes(stringSerializer.serialize(null, "k1"));
+ final Bytes k2 = new Bytes(stringSerializer.serialize(null, "k2"));
+ final Bytes k3 = new Bytes(stringSerializer.serialize(null, "k3"));
+ rocksDBStore.put(k1, stringSerializer.serialize(null, "a"));
+ rocksDBStore.put(k2, stringSerializer.serialize(null, "b"));
+ rocksDBStore.commit(Map.of());
+
+ rocksDBStore.put(k3, stringSerializer.serialize(null, "c"));
+ rocksDBStore.put(k1, stringSerializer.serialize(null, "a2"));
+
+ final ReadOnlyKeyValueStore<Bytes, byte[]> uncommitted =
rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED);
+ final ReadOnlyKeyValueStore<Bytes, byte[]> committed =
rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED);
+
+ final List<KeyValue<String, String>> uncommittedReverseAll;
+ try (KeyValueIterator<Bytes, byte[]> it = uncommitted.reverseAll()) {
+ uncommittedReverseAll = getDeserializedList(it);
+ }
+ final List<KeyValue<String, String>> committedReverseAll;
+ try (KeyValueIterator<Bytes, byte[]> it = committed.reverseAll()) {
+ committedReverseAll = getDeserializedList(it);
+ }
+ assertEquals(List.of(KeyValue.pair("k3", "c"), KeyValue.pair("k2",
"b"), KeyValue.pair("k1", "a2")), uncommittedReverseAll);
+ assertEquals(List.of(KeyValue.pair("k2", "b"), KeyValue.pair("k1",
"a")), committedReverseAll);
+
+ final List<KeyValue<String, String>> uncommittedReverseRange;
+ try (KeyValueIterator<Bytes, byte[]> it = uncommitted.reverseRange(k1,
k3)) {
+ uncommittedReverseRange = getDeserializedList(it);
+ }
+ final List<KeyValue<String, String>> committedReverseRange;
+ try (KeyValueIterator<Bytes, byte[]> it = committed.reverseRange(k1,
k3)) {
+ committedReverseRange = getDeserializedList(it);
+ }
+ assertEquals(List.of(KeyValue.pair("k3", "c"), KeyValue.pair("k2",
"b"), KeyValue.pair("k1", "a2")), uncommittedReverseRange);
+ assertEquals(List.of(KeyValue.pair("k2", "b"), KeyValue.pair("k1",
"a")), committedReverseRange);
+ }
+
+ @Test
+ public void
readOnlyReverseRangeShouldReturnEmptyIteratorWhenFromIsGreaterThanTo() {
+ rocksDBStore.init(context, rocksDBStore);
+ final Bytes k1 = new Bytes(stringSerializer.serialize(null, "k1"));
+ final Bytes k2 = new Bytes(stringSerializer.serialize(null, "k2"));
+ rocksDBStore.put(k1, stringSerializer.serialize(null, "a"));
+ rocksDBStore.put(k2, stringSerializer.serialize(null, "b"));
+
+ try (KeyValueIterator<Bytes, byte[]> it =
+
rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED).reverseRange(k2, k1)) {
+ assertFalse(it.hasNext());
+ }
+ }
+
+ @Test
+ public void readOnlyPrefixScanShouldRespectIsolationLevel() {
+ rocksDBStore.close();
+ final InternalMockProcessorContext<?, ?> eosContext =
getTransactionalEOSProcessorContext(dir);
+ rocksDBStore = getRocksDBStore();
+ rocksDBStore.init(eosContext, rocksDBStore);
+
+ rocksDBStore.put(new Bytes(stringSerializer.serialize(null, "p-1")),
stringSerializer.serialize(null, "a"));
+ rocksDBStore.commit(Map.of());
+ rocksDBStore.put(new Bytes(stringSerializer.serialize(null, "p-2")),
stringSerializer.serialize(null, "b"));
+ rocksDBStore.put(new Bytes(stringSerializer.serialize(null, "q-1")),
stringSerializer.serialize(null, "z"));
+
+ final List<KeyValue<String, String>> uncommittedPrefix;
+ try (KeyValueIterator<Bytes, byte[]> it =
rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED)
+ .prefixScan("p-", stringSerializer)) {
+ uncommittedPrefix = getDeserializedList(it);
+ }
+ final List<KeyValue<String, String>> committedPrefix;
+ try (KeyValueIterator<Bytes, byte[]> it =
rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED)
+ .prefixScan("p-", stringSerializer)) {
+ committedPrefix = getDeserializedList(it);
+ }
+ assertEquals(List.of(KeyValue.pair("p-1", "a"), KeyValue.pair("p-2",
"b")), uncommittedPrefix);
+ assertEquals(List.of(KeyValue.pair("p-1", "a")), committedPrefix);
+ }
+
+ @Test
+ public void
readOnlyOnNonTransactionalStoreShouldBehaveIdenticallyAcrossLevels() {
+ rocksDBStore.init(context, rocksDBStore);
+ final Bytes key = new Bytes(stringSerializer.serialize(null, "k"));
+ rocksDBStore.put(key, stringSerializer.serialize(null, "v"));
+
+ assertEquals("v", stringDeserializer.deserialize(null,
+ rocksDBStore.readOnly(IsolationLevel.READ_UNCOMMITTED).get(key)));
+ assertEquals("v", stringDeserializer.deserialize(null,
+ rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED).get(key)));
+ }
+
public static class TestingBloomFilterRocksDBConfigSetter implements
RocksDBConfigSetter {
static boolean bloomFiltersSet;
