chia7712 commented on code in PR #22643:
URL: https://github.com/apache/kafka/pull/22643#discussion_r3456780926
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -175,31 +178,72 @@ ManagedKeyValueIterator<Bytes, byte[]>
newBaseIterator(final Bytes from, final B
return newBaseIterator(cfHandle, from, to, forward, toInclusive);
}
+ @Override
+ ManagedKeyValueIterator<Bytes, byte[]> newBaseSnapshotIterator(final Bytes
from, final Bytes to,
+ final
boolean forward, final boolean toInclusive) {
+ return newBaseSnapshotIterator(cfHandle, from, to, forward,
toInclusive);
+ }
+
private ManagedKeyValueIterator<Bytes, byte[]> newBaseIterator(final
ColumnFamilyHandle cf,
final Bytes
from, final Bytes to,
final
boolean forward, final boolean toInclusive) {
final RocksIterator rocksIterator = db.newIterator(cf);
- final ManagedKeyValueIterator<Bytes, byte[]> iter;
+ final ManagedKeyValueIterator<Bytes, byte[]> iter =
+ buildBaseIterator(rocksIterator, from, to, forward, toInclusive);
+ // RocksDbIterator requires onClose to be set before close() is called.
+ // Since this iterator is used internally by StagedMergeIterator (not
+ // tracked by RocksDBStore's open-iterator set), use a no-op callback.
+ iter.onClose(() -> { });
Review Comment:
Should we move this to `buildBaseIterator` to ensure the returned
`RocksDbIterator` is all-set?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -175,31 +178,72 @@ ManagedKeyValueIterator<Bytes, byte[]>
newBaseIterator(final Bytes from, final B
return newBaseIterator(cfHandle, from, to, forward, toInclusive);
}
+ @Override
+ ManagedKeyValueIterator<Bytes, byte[]> newBaseSnapshotIterator(final Bytes
from, final Bytes to,
+ final
boolean forward, final boolean toInclusive) {
+ return newBaseSnapshotIterator(cfHandle, from, to, forward,
toInclusive);
+ }
+
private ManagedKeyValueIterator<Bytes, byte[]> newBaseIterator(final
ColumnFamilyHandle cf,
final Bytes
from, final Bytes to,
final
boolean forward, final boolean toInclusive) {
final RocksIterator rocksIterator = db.newIterator(cf);
- final ManagedKeyValueIterator<Bytes, byte[]> iter;
+ final ManagedKeyValueIterator<Bytes, byte[]> iter =
+ buildBaseIterator(rocksIterator, from, to, forward, toInclusive);
+ // RocksDbIterator requires onClose to be set before close() is called.
+ // Since this iterator is used internally by StagedMergeIterator (not
+ // tracked by RocksDBStore's open-iterator set), use a no-op callback.
+ iter.onClose(() -> { });
+ return maybeWrapRangeTombstones(iter);
+ }
+
+ private ManagedKeyValueIterator<Bytes, byte[]>
newBaseSnapshotIterator(final ColumnFamilyHandle cf,
+
final Bytes from, final Bytes to,
+
final boolean forward, final boolean toInclusive) {
+ final Snapshot snapshot = db.getSnapshot();
Review Comment:
Should we ensure the `snapshot` gets closed?
--
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]
