nicktelford commented on code in PR #23999:
URL: https://github.com/apache/kafka/pull/23999#discussion_r4425969599
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -286,21 +294,58 @@ private ManagedKeyValueIterator<Bytes, byte[]>
newBaseIterator(final ColumnFamil
return new RangeTombstoneFilterIterator(iter, rangeTombstones);
}
+ @Override
+ void prepareFlush() {
+ preparedBatch = new WriteBatch();
+ try {
+ for (final PendingRangeDelete delete : pendingRangeDeletes) {
+ preparedBatch.deleteRange(delete.cf, delete.from.get(),
delete.to.get());
+ }
+ addPoints(cfHandle, pendingWrites);
+ for (final Map.Entry<ColumnFamilyHandle, NavigableMap<Bytes,
Optional<byte[]>>> cfWrites : otherCfWrites.entrySet()) {
+ addPoints(cfWrites.getKey(), cfWrites.getValue());
+ }
+ } catch (final RocksDBException e) {
+ preparedBatch.close();
+ preparedBatch = null;
+ throw new ProcessorStateException("Error preparing commit batch
for store " + storeName, e);
+ }
+ }
+
+ private void addPoints(final ColumnFamilyHandle cf,
+ final NavigableMap<Bytes, Optional<byte[]>> writes)
throws RocksDBException {
+ for (final Map.Entry<Bytes, Optional<byte[]>> entry :
writes.entrySet()) {
+ if (entry.getValue().isPresent()) {
+ preparedBatch.put(cf, entry.getKey().get(),
entry.getValue().get());
Review Comment:
Isn't this an issue with the existing CachingKeyValueStore too? Keys and
values are not copied on `put`, so could be hypothetically changed before cache
flush (which is triggered just before commit).
In practice, I don't think this is something we should defend against. The
cost of copying all key/value byte arrays (both in the transaction buffer, and
in the cache) would simply be too high, and would exist only to defend against
an extremely unlikely edge case.
--
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]
