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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
