aliehsaeedii commented on code in PR #14626: URL: https://github.com/apache/kafka/pull/14626#discussion_r1404613145
########## streams/src/main/java/org/apache/kafka/streams/state/internals/LogicalKeyValueSegment.java: ########## @@ -170,7 +173,30 @@ public boolean isOpen() { @Override public synchronized byte[] get(final Bytes key) { - return physicalStore.get(prefixKeyFormatter.addPrefix(key)); + return get(key, Optional.empty()); + } + + public synchronized byte[] get(final Bytes key, final Snapshot snapshot) { + return get(key, Optional.of(snapshot)); + } + + private synchronized byte[] get(final Bytes key, final Optional<Snapshot> snapshot) { + if (snapshot.isPresent()) { + try (ReadOptions readOptions = new ReadOptions()) { + readOptions.setSnapshot(snapshot.get()); + return physicalStore.get(prefixKeyFormatter.addPrefix(key), readOptions); + } + } else { + return physicalStore.get(prefixKeyFormatter.addPrefix(key)); + } + } + + public Snapshot getSnapshot() { + return physicalStore.db.getSnapshot(); Review Comment: > I think we should push this into `RocksDBStore` class, ie, call `physicalStore.getSnapshot()` here, and extend `RocksDBStore` to track all open snapshots (similar to how we track open iterators and release all open snapshot if the store is closed). Since "Snapshots do not persist across DB restarts ([ref](https://github.com/facebook/rocksdb/wiki/Snapshot))", I think we do not need track closing them when closing the store. -- 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-unsubscr...@kafka.apache.org For queries about this service, please contact Infrastructure at: us...@infra.apache.org