You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2026/06/26 08:27:44 UTC
[PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
nicktelford opened a new pull request, #22682:
URL: https://github.com/apache/kafka/pull/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.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #22682:
URL: https://github.com/apache/kafka/pull/22682#discussion_r3483030289
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java:
##########
@@ -60,26 +61,53 @@ 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.viewAccessor(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;
}
@Override
public synchronized void destroy() {
Review Comment:
I may be missing something but could this cause a problem on closing - I haven't read the code so just asking here.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBVersionedStore.java:
##########
@@ -272,31 +282,123 @@ public VersionedRecord<byte[]> get(final Bytes key, final long asOfTimestamp) {
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) {
Review Comment:
nit: parameter alignment
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #22682:
URL: https://github.com/apache/kafka/pull/22682#discussion_r3491368537
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java:
##########
@@ -60,26 +61,53 @@ 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.viewAccessor(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;
}
@Override
public synchronized void destroy() {
Review Comment:
`destroy` isn't called by Interactive Queries or read-only views, so it shouldn't be an issue. Unless that's not what you were asking?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on PR #22682:
URL: https://github.com/apache/kafka/pull/22682#issuecomment-4813141985
@nicktelford - PR has checkstyle issues
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #22682:
URL: https://github.com/apache/kafka/pull/22682#issuecomment-4807813353
@bbejeck
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-20500: Add isolation-level reads to versioned stores [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #22682:
URL: https://github.com/apache/kafka/pull/22682#discussion_r3494680272
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java:
##########
@@ -60,26 +61,53 @@ 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.viewAccessor(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;
}
@Override
public synchronized void destroy() {
Review Comment:
Nope that's good, just clarifying
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org