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
We need to make the distinction between a tombstone and "no staged write",
because when the buffer has no staged write, we delegate reads to the
underlying RocksDB store (to read the last committed value), but when we have a
staged tombstone we _immediately_ return `null`, because that tombstone
overrides any committed value in RocksDB.
If you don't like the spurious IntelliJ warnings, and the spotbugs
exclusion, I can always add an `enum` to capture these 3 states, but I didn't
think it was worth the extra API surface.
--
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]
