nicktelford commented on code in PR #22625:
URL: https://github.com/apache/kafka/pull/22625#discussion_r3451256496
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -1034,6 +1050,115 @@ public void close() {
}
+ static class TransactionalDBAccessor implements DBAccessor {
+
+ private final DBAccessor underlying;
+ private final RocksDBTransactionBuffer buffer;
+ private final ColumnFamilyHandle cfHandle;
+
+ TransactionalDBAccessor(final DBAccessor underlying,
+ final RocksDB db,
+ final ColumnFamilyHandle cfHandle,
+ final WriteOptions wOptions,
+ final String storeName) {
+ this.underlying = underlying;
+ this.cfHandle = cfHandle;
+ this.buffer = new RocksDBTransactionBuffer(db, cfHandle, wOptions,
storeName);
+ }
+
+ @Override
+ public byte[] get(final ColumnFamilyHandle columnFamily, final byte[]
key) throws RocksDBException {
+ if (columnFamily.equals(cfHandle)) {
+ final java.util.Optional<byte[]> staged =
buffer.get(Bytes.wrap(key));
+ if (staged != null) {
+ return staged.orElse(null);
+ }
+ }
+ return underlying.get(columnFamily, key);
+ }
+
+ @Override
+ public byte[] get(final ColumnFamilyHandle columnFamily, final
ReadOptions readOptions, final byte[] key) throws RocksDBException {
+ if (columnFamily.equals(cfHandle)) {
+ final java.util.Optional<byte[]> staged =
buffer.get(Bytes.wrap(key));
Review Comment:
TL;DR: tombstones
`TransactionBuffer#get` has 3 possible return states:
* `Optional.of` some value - the value actually staged in the buffer for the
key
* `Optional.empty()` - the staged value is a tombstone that needs to be
written to the RocksDB store on-commit.
* `null` - no value or tombstone has been staged for the key
--
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]
