Copilot commented on code in PR #23999:
URL: https://github.com/apache/kafka/pull/23999#discussion_r4424664041
##########
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:
With `Serdes.ByteArray()` and caching disabled, modifying an input array or
a value returned by a staged read before commit now changes what RocksDB
persists without another `put`; the previous WriteBatch captured the bytes when
staged. Snapshot keys, values, and range endpoints when staging, and keep reads
and iterators from exposing those snapshots.
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/AbstractTransactionBuffer.java:
##########
@@ -100,9 +103,11 @@ abstract ManagedKeyValueIterator<K, byte[]>
newBaseSnapshotIterator(K from, K to
public void stage(final K key, final byte[] value) {
snapshotLock.writeLock().lock();
try {
- pendingWrites.put(key, Optional.ofNullable(value));
+ final Optional<byte[]> previous = pendingWrites.put(key,
Optional.ofNullable(value));
+ if (previous != null) {
+ pendingWritesBytes -= estimateKeySize(key) + previous.map(v ->
v.length).orElse(0);
+ }
pendingWritesBytes += estimateKeySize(key) + (value != null ?
value.length : 0);
Review Comment:
Counting only key/value bytes makes an empty-key write with an empty value
or tombstone report zero, so a write from `Processor.init()` can fail to set
`commitNeeded` in `StreamTask.completeRestoration()` and remain uncommitted
while the task is idle. For short keys and tombstones, this also understates
the native batch size used to reserve commit memory in both stream and global
threads. Include conservative batch-record overhead in the accounting, covering
primary and other column-family writes, range deletes, and the header of a
nonempty batch.
--
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]
