You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "nicktelford (via GitHub)" <gi...@apache.org> on 2026/10/28 16:44:36 UTC
[PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
nicktelford opened a new pull request, #23999:
URL: https://github.com/apache/kafka/pull/23999
`RocksDBTransactionBuffer` writes every staged write twice: to `pendingWrites` on the heap, which serves reads, and to a native RocksDB `WriteBatch`. The `WriteBatch` holds native memory for the whole transaction, and the buffer changes it while it holds the write lock.
The buffer now stages writes on the heap only. Writes to other column families and range deletes also stay on the heap until the commit. At commit, `prepareFlush()` builds the `WriteBatch` from the staged state outside the write lock, and `flushToBase()` writes it to RocksDB under the write lock. A JMH model of the two designs measured writes 2 to 7 times faster, and commits 8 to 12 times faster when the transaction overwrites keys. A commit with no overwrites and 1000-byte values is up to 28% slower.
Because no buffer writes to a backend at stage time now, the `stageToBackend` hook of `AbstractTransactionBuffer` has no function. This change removes it and all of its implementations.
The uncommitted-bytes estimate also counted a key once for each write in a transaction, so a key written many times could cause an early commit with little staged data. `AbstractTransactionBuffer.stage` now subtracts the size of the previous entry for the key before it adds the new entry. `RocksDBTransactionBuffer` uses this shared accounting for all of its staged data.
`statestore.uncommitted.max.bytes` now counts each staged write once. Before, it counted the heap copy and the `WriteBatch` copy. The commit still needs a native `WriteBatch`, but only for one store at a time, because stores commit one after the other. The extra memory at commit is therefore at most the uncommitted bytes of the largest store. A thread now commits early when its uncommitted bytes plus the uncommitted bytes of its largest store exceed its share of the limit. The global thread uses the same check. The limit is thus again a bound on the memory at commit.
The check uses the largest value of `StateStore.approximateNumUncommittedBytes()`, so it needs no change to a public interface. In-memory stores need no extra memory at commit, so applications with only in-memory stores commit early after less data than in 4.4. Applications with more than one transactional RocksDB store can buffer more data before an early commit. The 4.5.0 upgrade guide and the documentation of the config record this. The default stays at 64 MB.
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "bbejeck (via GitHub) via jira" <ji...@kafka.apache.org>.
bbejeck commented on PR #23999:
URL: https://github.com/apache/kafka/pull/23999#issuecomment-6533699533
[System tests passed](https://confluent-open-source-kafka-branch-builder-system-test-results.s3-us-west-2.amazonaws.com/trunk/2026-10-30--001.c9d3bed3-a3f3-4a0c-a9df-a8fb6f0b5d62--1793391408--nicktelford--KAFKA-21230-lazy-writebatch--7d43962b3f/report.html)
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on PR #23999:
URL: https://github.com/apache/kafka/pull/23999#issuecomment-6527939715
@bbejeck I've responded to both Copilot review comments: I believe neither should be addressed in this PR. Let me know if you feel differently.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "Copilot (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "bbejeck (via GitHub)" <gi...@apache.org>.
bbejeck commented on code in PR #23999:
URL: https://github.com/apache/kafka/pull/23999#discussion_r4425901665
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -135,16 +139,24 @@ void stageAll(final Runnable stagingOperations) {
/**
* Stages a range deletion for an explicit column family. Updates the shared
* {@code pendingWrites} and {@code rangeTombstones} so iterators opened before
- * commit hide the deleted range, and appends the range delete to the shared
- * {@link WriteBatch} under {@code cf}.
+ * commit hide the deleted range, and records the range delete for {@code cf} so
Review Comment:
```suggestion
* commits hide the deleted range, and records the range delete for {@code cf} so
```
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -135,16 +139,24 @@ void stageAll(final Runnable stagingOperations) {
/**
* Stages a range deletion for an explicit column family. Updates the shared
* {@code pendingWrites} and {@code rangeTombstones} so iterators opened before
- * commit hide the deleted range, and appends the range delete to the shared
- * {@link WriteBatch} under {@code cf}.
+ * commit hide the deleted range, and records the range delete for {@code cf} so
+ * {@link #prepareFlush()} adds it to the {@link WriteBatch}.
*/
void stageDeleteRange(final ColumnFamilyHandle cf, final Bytes from, final Bytes to) {
snapshotLock.writeLock().lock();
try {
// Multi-node subMap().clear() on the (non-thread-safe) TreeMap staging buffer must hold
// the write lock; doing it together with the rangeTombstones swap also makes the two
// updates atomic with respect to a non-owner snapshot scan.
- pendingWrites.subMap(from, true, to, false).clear();
+ final NavigableMap<Bytes, Optional<byte[]>> staged =
+ cf == cfHandle ? pendingWrites : otherCfWrites.get(cf);
+ if (staged != null) {
Review Comment:
I get that we need to check for `null` but I'm wondering - The fact that an expected cf is `null` seems somewhat noteworthy - worth a log warning statement? Maybe I'm making somethign out of nothing.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "nicktelford (via GitHub)" <gi...@apache.org>.
nicktelford commented on code in PR #23999:
URL: https://github.com/apache/kafka/pull/23999#discussion_r4425984237
##########
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:
I believe this issue is pre-existing, and not newly introduced in this PR.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-21230: Build RocksDB WriteBatch at commit [kafka]
Posted by "nicktelford (via GitHub) via jira" <ji...@kafka.apache.org>.
nicktelford commented on code in PR #23999:
URL: https://github.com/apache/kafka/pull/23999#discussion_r4427957248
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -135,16 +139,24 @@ void stageAll(final Runnable stagingOperations) {
/**
* Stages a range deletion for an explicit column family. Updates the shared
* {@code pendingWrites} and {@code rangeTombstones} so iterators opened before
- * commit hide the deleted range, and appends the range delete to the shared
- * {@link WriteBatch} under {@code cf}.
+ * commit hide the deleted range, and records the range delete for {@code cf} so
+ * {@link #prepareFlush()} adds it to the {@link WriteBatch}.
*/
void stageDeleteRange(final ColumnFamilyHandle cf, final Bytes from, final Bytes to) {
snapshotLock.writeLock().lock();
try {
// Multi-node subMap().clear() on the (non-thread-safe) TreeMap staging buffer must hold
// the write lock; doing it together with the rangeTombstones swap also makes the two
// updates atomic with respect to a non-owner snapshot scan.
- pendingWrites.subMap(from, true, to, false).clear();
+ final NavigableMap<Bytes, Optional<byte[]>> staged =
+ cf == cfHandle ? pendingWrites : otherCfWrites.get(cf);
+ if (staged != null) {
Review Comment:
I believe `staged` can be `null` if no other writes have yet been staged for this `ColumnFamilyHandle`. The `otherCfWrites` maps are lazily initialized (see `stage`, line 109). This branch appears to wipe any "pending writes" in the range of the range delete. If no writes to this CF have been staged, then there's nothing to wipe, so it can be skipped.
--
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-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org