KAFKA-20578: Initial producer incremental allocation for uncompressed data by lianetm · Pull Request #22654 · apache/kafka

lianetm · 2026-06-23T18:34:14Z

Initial partial implementation for KIP-1332 (producer incremental
allocation strategy).

This PR includes:

new producer config for allocation strategy (incremental/full)
initial implementation of the incremental strategy: supports
uncompressed data only (no growth support), extra-copy on send (linked
chunks are flattened into a new buffer)
unit and integration tests (running existing Producer integration
tests with the new strategy + new ones)

Support for compressed data and network layer improvements will come in
follow-up PRs.

Reviewers: Jun Rao junrao@gmail.com, Ken Huang s7133700@gmail.com,
Chia-Ping Tsai chia7712@gmail.com

junrao

@lianetm : Thanks for the PR. Made a pass of the non-testing files. Left a few comments.

junrao · 2026-06-29T20:48:51Z

                                        Importance.MEDIUM,
                                        CommonClientConfigs.CLIENT_DNS_LOOKUP_DOC)
                                .define(BUFFER_MEMORY_CONFIG, Type.LONG, 32 * 1024 * 1024L, atLeast(0L), Importance.HIGH, BUFFER_MEMORY_DOC)
+                                .define(BUFFER_MEMORY_ALLOCATION_STRATEGY_CONFIG,


Should we mark this as internal to prevent it from leaking into 4.4 before the feature is fully implemented?

Yes, done, and created https://issues.apache.org/jira/browse/KAFKA-20763 to publish it once the feature is complete

junrao · 2026-06-30T00:10:36Z

+                    // pool memory by completing batches). On exhaustion, close the batch (making it
+                    // drainable) and fall through to the blocking first-record path next iteration.
+                    try {
+                        extensionChunks = chunkedFree.allocateChunks(extensionBytes, 0L);


Why do we need to do a special non-blocking call? In the existing logic, if an allocation request is blocked, all existing ProducerBatches become immediately drainable.

Not strictly needed, agreed, it was just to fail fast and avoid what seemed like a "waisted" wait here, exactly because of the fact you mention, that blocking would make all open batches drainable, including this one we want to extend (and this is the trick I was trying to avoid, updated the comment that was misleading).

So with a blocking call, even if we get the pool memory we need in time, the batch would probably be gone/drained, and we would need to start a new batch anyways (so opting for non-blocking just to go straight to the new-batch phase if pool is exhausted when extending).

That being said, there is a case where blocking here would seem better, that's if the pool is exhausted at this point but it frees-up before the batch we're extending is drained. But that one didn't seem common in practice given that the sender drains batches before polling (so by the time memory frees up from a request that compelted, the batch we're waiting to extend would be drained already). Makes sense?

junrao · 2026-06-30T00:12:57Z

+                                last.closeForRecordAppends();
+                            }
+                        }
+                        continue;


It may take a bit of time for the closed batches to be drained. If we continue here, it seems that the client will just busy-loop until some batches are drained and some free space becomes available in buffer pool?

even without the drain the closeForAppends should be enough to make that the next iteration find that the stream is closed => isFull => !hasRoom so starting a new batch (blocking, no busy loop). Makes sense? Added a test testExhaustedExtensionFallsBackToBlockingNewBatchPath to show the behaviour.

junrao · 2026-06-30T20:49:43Z

+     * them, and retries. Otherwise defers to the parent.
+     */
+    @Override
+    protected RecordAppendResult tryAppend(long timestamp, byte[] key, byte[] value, Header[] headers,


It's a bit awkward to have a return value of null and RecordAppendResult.needsExtension. Could we introduce a non-null value to indicate the batch is full?

Agreed, done. Introduced needsNewBatch (aligned with needsBufferExtension)

junrao · 2026-06-30T20:57:38Z

+                        // ProducerBatch), which can't take extension chunks. Only attach to a
+                        // writable chunked batch; otherwise refund the chunks and re-evaluate.
+                        if (last instanceof ChunkedProducerBatch && last.isWritable()) {
+                            ((ChunkedProducerBatch) last).addBuffers(extensionChunks);


I guess two concurrent clients could add buffers exceeding the batch size? Those buffers won't be used, but can only be freed after the batch is drained.

Correct, there could be over-reservation at this point with concurrent appends. As for how the extra buffers would be used/released, I see the following cases:

used right away: next loop iteration -> the loser's tryAppend right after this line fails with needsExtension (the winner already took the free capacity that was used to calculate this extension). Then the loser loops and tries again (recomputes the extension needed, so it uses what it had added concurrently + more)

left unused, freed when batch drains (case you mentioned) -> if the loser's record can't land in this batch (batch closes and loser's record goes into new batch).

Added 2 tests to cover this: testConcurrentExtensionRaceDoesNotOverflowChunkedStream and testBatchSizeLimitRespectedDespiteOverReservation

Here is a related case. Two concurrent producer each allocated its own extensionChunks. It's possible that the first producer's allocation is enough to fit the records from both producer. It would be useful to document that this logic optimizes for the common case, but can lead to slightly temporary over allocation.

Yes, agreed, added a comment describing it.

junrao · 2026-06-30T23:01:33Z

+                long remainingBytes = memoryRequired - (long) pooled.size() * chunkSize;
+                if (remainingBytes > 0) {
+                    // remainingBytes <= memoryRequired <= totalMemory (validated above), so the int cast is safe.
+                    freeUp((int) remainingBytes);


This is a no-op since all pooled chunks have been used if we reach here.

agreed, removed.

junrao · 2026-07-01T00:48:52Z

+                    pooled.add(free.pollFirst());
+                long remainingBytes = memoryRequired - (long) pooled.size() * chunkSize;
+                if (remainingBytes > 0) {
+                    // remainingBytes <= memoryRequired <= totalMemory (validated above), so the int cast is safe.


Why is remainingBytes guaranteed to be an int? memoryRequired could be larger than int and pooled.size() initially could be 0.

yup, good catch, seemed safe here but it's not because of how memoryRequired is rounded up higher up. This code here does not exist anymore though so no changes (the other place where a similar unsafe op was being done to calculate stillNeeded went away too, removed with the comment/fix below)

junrao · 2026-07-01T16:11:40Z

+                        }
+                        long stillNeeded = memoryRequired - (long) pooled.size() * chunkSize - accumulated;
+                        if (stillNeeded > 0) {
+                            freeUp((int) stillNeeded);


This may be ok, but it's a bit weird to free up the chunks only to be reallocated again. Here is an alternative that doesn't require a freeup() call.

// Reuse pooled chunks first. If a reused chunk covers a slot we already reserved as // raw bytes in an earlier iteration, hand that raw reservation back to the pool. while (pooled.size() < numChunks && !free.isEmpty()) { pooled.add(free.pollFirst()); if (accumulated >= chunkSize) { // accumulated is always chunk-aligned here accumulated -= chunkSize; this.nonPooledAvailableMemory += chunkSize; // refund → available to other waiters } } // Reserve raw memory for any still-uncovered chunks, in whole chunks. while (pooled.size() + (int)(accumulated / chunkSize) < numChunks && this.nonPooledAvailableMemory >= chunkSize) { this.nonPooledAvailableMemory -= chunkSize; accumulated += chunkSize; }

nice, makes sense, done.

junrao · 2026-07-01T16:33:06Z

+
+                        // Take pool chunks first, then reserve non-pool bytes for the remainder.
+                        while (pooled.size() < numChunks
+                                && (long) (pooled.size() + 1) * chunkSize + accumulated <= memoryRequired


The first condition is redundant, given the second one.

(pooled+1)*chunkSize + accumulated ≤ memoryRequired ⇒ (pooled+1)*chunkSize ≤ memoryRequired − accumulated ≤ memoryRequired = numChunks*chunkSize ⇒ pooled+1 ≤ numChunks ⇒ pooled < numChunks

Agreed (the condition changed though, with the comment above, and does not have the redundancy anymore, so no changes directly for this)

junrao · 2026-07-01T17:19:09Z

+                    // On failure (timeout / close / interrupt), refund the non-pool bytes taken.
+                    // Pool chunks already in `pooled` are returned to `free` separately by the
+                    // outer catch.
+                    this.nonPooledAvailableMemory += accumulated;


Could we return accumulated and pooled chunks in the same place? For example, we can set a flag like allocationCompleted to replace accumulated = 0. Then we can free both accumulated and pooled chunks if the flag is not set.

Done. Introduced the allocationCompleted flag and now all refunds happen together in a finally, much better indeed.

lianetm · 2026-07-15T20:35:25Z

Thanks for the comments @junrao ! All addressed (just some changes pending the parallel refactoring, details in the answers above)

chia7712 · 2026-07-16T21:49:28Z

+     * @throws BufferExhaustedException if the request can't be satisfied within {@code maxTimeToBlockMs}
+     * @throws KafkaException           if the pool is closed during the wait
+     */
+    public List<ByteBuffer> allocateChunks(int totalSize, long maxTimeToBlockMs) throws InterruptedException {


Out of curiosity, have you considered moving allocateChunks into BufferPool instead of having specific ChunkedBufferPool? Also, allocate seems to be a variant of allocateChunks which would simplify the codebase.

interesting, I moved the allocateChunks into BufferPool and it's a good simplification indeed. I also extracted the common logic that applies to allocate and allocateChunks (awaitMemory), but didn't go for a full merge of the 2 because even though similar (the common logic extracted), they have some differences and made the fully-consolidated approach tricky (e.g., the nature of what they reserve, full chunks vs bytes, metrics used differently). Makes sense?

junrao

@lianetm : Thanks for the updated PR. A few more comments.

junrao · 2026-07-17T21:22:02Z

+                        continue;
+                    }
+                    if (appendResult.needsNewBatch())
+                        throw new IllegalStateException("appendNewBatch must not return a needsNewBatch result");


Could we do this check right after we get appendResult back?

junrao · 2026-07-17T21:32:55Z

+         */
+        public enum Outcome { APPENDED, NEEDS_BUFFER_EXTENSION, NEEDS_NEW_BATCH }
+
+        public final Outcome outcome;


Should this be private?

yes, both indeed, done + comment on what the callers invoke.

junrao · 2026-07-17T21:36:22Z

     * @param nowMs The current time, in milliseconds
+     * @return the append result, which never has {@code needsNewBatch=true}: the method either
+     *         propagates a non-{@code needsNewBatch} result from an open batch created concurrently
+     *         (a success, or — incremental strategy — a {@code needsBufferExtension} signal), or it


This comment is not very clear to me. a success should return appended, not needsBufferExtension, right?

yes, that was the intention on the comment: success (appended) OR needsBufferExtension. Updated it to make it clearer.

junrao · 2026-07-17T21:45:31Z

+     */
+    public void addBuffers(List<ByteBuffer> newChunks) {
+        ensureNotDeallocated();
+        chunks.addAll(newChunks);


Should we call ensureWritable() here too?

junrao · 2026-07-17T22:03:49Z

+        }
+        chunks.clear();
+        currentChunk = null;
+        currentChunkIndex = 0;


0 is a valid index. Should we set it to -1?

junrao · 2026-07-17T22:04:37Z

+
+    @Override
+    public void ensureRemaining(int remainingBytesRequired) {
+        ensureNotDeallocated();


Should we call this in limit() and initialCapacity() too?

makes sense (done + tests)

junrao · 2026-07-17T22:16:37Z

+    @Override
+    public ByteBuffer buffer() {
+        ensureNotDeallocated();
+        finalized = true;


It's probably better to set this when close() is called. We can then rename it to closed.

agreed, done

junrao · 2026-07-17T22:41:13Z

+                    // so bufferStream was allocated (this iteration or carried from a prior one).
+                    if (bufferStream == null)
+                        throw new IllegalStateException("needsNewBatch path reached without an allocated buffer stream");
+                    int firstRecordSize = AbstractRecords.estimateSizeInBytesUpperBound(


This code is different from the code when the new batch is allocated. Could we remember the first record size when the new batch is allocated and reuse it here?

int recordUncompressed = AbstractRecords.recordSizeUpperBound( RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), key, value, headers); int size = MemoryRecordsBuilder.estimatedBytesWritten( RecordBatch.CURRENT_MAGIC_VALUE, compression.type(), CompressionRatioEstimator.estimation(topic, compression.type()), recordUncompressed);

yes, done (will probably need review once compression lands though, just added a TODO for now and we can discuss once there)

lianetm · 2026-07-20T17:34:12Z

Thanks @junrao and @chia7712 for the reviews! All comments addressed.

junrao

@lianetm : Thanks for the updated PR. A few more comments. Also, there is code duplicated between RecordAccumulator and ChunkedRecordAccumulator. Claude suggested the following two alternatives. Is that worth doing?

Alternative 1: Keep ChunkedRecordAccumulator as subclass.
  // base RecordAccumulator (the single loop), pseudo-hooks:                                                                                                                                                                                                                                                                
                                                                                                                                                                                                                                                                                                                            
  // 1) allocate storage for a NEW batch, return the builder supplier (+ a handle to release on cleanup)                                                                                                                                                                                                                    
  protected BatchBuilderHandle allocateNewBatchBuilder(topic, part, record, maxTimeToBlock) { … }                                                                                                                                                                                                                           
  //   Full:    ByteBuffer via free.allocate → () -> MemoryRecords.builder(buffer, …)                                                                                                                                                                                                                                       
  //   Chunked: chunks via allocateChunks → new ChunkedByteBufferOutputStream → () -> chunkedRecordsBuilder(stream, firstRecordSize)                                                                                                                                                                                        
                                                                                                                                                                                                                                                                                                                            
  // 2) handle the NEEDS_BUFFER_EXTENSION outcome (allocate off-lock, re-lock, attach, retry)                                                                                                                                                                                                                               
  protected void extendOpenBatch(dq, extensionBytesNeeded, maxTimeToBlock) { … }                                                                                                                                                                                                                                            
  //   Full:    unreachable — tryAppend never returns NEEDS_BUFFER_EXTENSION (fixed buffer == writeLimit)                                                                                                                                                                                                                   
  //   Chunked: allocateChunks(0L) → addBuffers → retry / close-and-fall-through on exhaustion                                                                                                                                                                                                                              
                                                                                                                                                                                                                                                                                                                            
  The single loop then does:                                                                                                                                                                                                                                                                                                
  while (true) {                                                                                                                                                                                                                                                                                                            
      ... pick partition, get dq ...                                                                                                                                                                                                                                                                                        
      synchronized (dq) {                                                                                                                                                                                                                                                                                                   
          if (partitionChanged(...)) continue;                                                                                                                                                                                                                                                                              
          RecordAppendResult r = tryAppend(...);                                                                                                                                                                                                                                                                            
          switch (r.outcome) {                                                                                                                                                                                                                                                                                              
              case APPENDED:            updatePartitionInfo(...); return r;                                                                                                                                                                                                                                                 
              case NEEDS_BUFFER_EXTENSION: /* release lock */ extendOpenBatch(...); continue;                                                                                                                                                                                                                               
              case NEEDS_NEW_BATCH:     ... allocateNewBatchBuilder(...) → appendNewBatch(...) ...                                                                                                                                                                                                                          
          }                                                                                                                                                                                                                                                                                                                 
      }                                                                                                                                                                                                                                                                                                                     
  }                                                                                                                                                                                                                                                                                                                         

Alternative 2:  Composition over inheritance (cleaner, bigger change): instead of ChunkedRecordAccumulator extends RecordAccumulator, inject a 
BatchAllocationStrategy (Full vs Chunked) into a single RecordAccumulator. One accumulator, one loop, strategy object handles storage + extension.

junrao · 2026-07-21T16:39:50Z

+     * @throws BufferExhaustedException if the request can't be satisfied within {@code maxTimeToBlockMs}
+     * @throws KafkaException           if the pool is closed during the wait
+     */
+    public List<ByteBuffer> allocateChunks(int totalSize, long maxTimeToBlockMs) throws InterruptedException {


I guess the caller is expected to use either allocate() or allocateChunks(). Should we prevent the caller from mixing the two calls?

yes. I added an allocation mode to the pool, then we can easily enforce not only that allocate and allocateChunks are not mixed, but also that they are used with the right intention. Thoughts?

junrao · 2026-07-21T17:19:46Z

-     *  resources like compression buffers. The batch will be fully closed (ie. the record batch headers will be written
-     *  and memory records built) in one of the following cases (whichever comes first): right before send,
-     *  if it is expired, or when the producer is closed.
+     *  If it is full (or absent), we return {@link RecordAppendResult#NEEDS_NEW_BATCH} and a new batch is created.


It would be useful to document what types of RecordAppendResult this method could return.

sure, done (here and on the chunk override)

junrao · 2026-07-21T17:24:34Z

+     * @return the append result, which is never {@code needsNewBatch}. It is either {@code appended}
+     *         — the record was appended, whether to a batch another thread created concurrently or to
+     *         the batch this method creates — or, in the incremental strategy, {@code needsBufferExtension}:
+     *         a concurrent appender created an extendable open batch, so the caller releases its


a concurrent appender created an extendable open batch => a concurrent appender created an extendable open batch, but the new record doesn't fit in.

agreed, clarified.

junrao · 2026-07-21T17:43:05Z

+                        throw e;
+                    }
+                    nowMs = time.milliseconds();
+                    bufferStream = new ChunkedByteBufferOutputStream(initialChunks, chunkedFree.poolableSize(), chunkedFree);


Could we put newBatchSize and bufferStream in a record since they should always be present together?

sure, makes sense (just that it cannot be in a record, java 11, but consolidated them with a class)

junrao · 2026-07-21T18:30:32Z

+     * Unlike the single-buffer batch — which must donate a fresh buffer because the network
+     * layer may still be reading the pooled one — a chunked batch's inflight bytes live in the
+     * separate flattened buffer (see {@link ChunkedByteBufferOutputStream#buffer()}), so it is
+     * safe to return the actual chunks to the pool here.


a chunked batch's inflight bytes live in the separate flattened buffer

This is true for now, but won't be true when we avoid flattening the buffer. Should we add a TODO to remind ourselves?

junrao · 2026-07-21T22:40:29Z

+     * appender racing the same batch.
+     */
+    private BufferPool poolMockingConcurrentChunkAllocation(int chunkSize, long totalMemory,
+                                                                   AtomicReference<ChunkedRecordAccumulator> injectAppendOnce,


indentation

junrao · 2026-07-21T23:31:17Z

+            // The original batch is far below its writeLimit, so isFull() can only be true via
+            // the closed append stream — i.e., it was closed for appends on the failed extension.
+            assertTrue(dq.peekFirst().isFull(), "original batch must be closed for appends");
+            assertNotNull(dq.peekFirst());


Should we do this one before the previous line?

sure, done (it was a dup I introduced to fix a warning from the IDE). Fixed a similar one too.

junrao · 2026-07-21T23:34:29Z

+
+            // Second record overflows the batch's chunk, so extension attempt fails (pool empty,
+            // recovered by closing the batch), then the new-batch acquire fails too (maxTimeToBlock
+            // = 0). The two failed acquires must count as a single dropped record.


The new batch acquire should have maxTimeToBlock != 0, right?

yes, right. Updated. The extension is the only one non-blocking in this case, then the new batch acquire is done with the maxTimeToBlock != 0

junrao · 2026-07-21T23:36:47Z

+     * Test that chunks are returned to the pool only at batch completion (deallocate), never at close.
+     */
+    @Test
+    public void testBatchCloseDoesNotDeallocateChunksPrematurely() throws Exception {


Where does it test the deallocation doesn't happen?

good catch, it was checking the records were still readable, but not directly checking the memory, fixed

junrao · 2026-07-21T23:40:56Z

+        assertNotNull(batch);
+        assertEquals(6, batch.recordCount);
+
+        // batch.close() flattens chunks and writes the header. If chunks under-allocated, this


Hmm, if chunks are under allocated, we should throw at append time, not close time, right?

yes, updated the comment (assertions were already aligned). No throw at close here, under-allocation would be an IllegalState on tryAppend.

lianetm · 2026-07-22T21:49:44Z

Thanks for the review @junrao , all comments addressed.

About the high level one (this) on the overlap between RecordAccumulator and ChunkedRecordAccumulator, totally agree. I refactored to extract and reuse different common bits for now (keeping both classes still, just common helpers, along the lines of the alternative 1). Compression will touch the core of this, so wonder if better to wait to see how those changes fit here, to see how much common surface we still have and then decide if continue with the helpers approach, taking them a step further, or go with some like the alt 2 approach. Makes sense? if so I would create a jira on me to review/dedup along with the compression changes.

junrao

@lianetm : Thanks for the updated PR. Just a few minor comments.

junrao · 2026-07-22T23:24:10Z

+     * strategy), {@link #CHUNKED} only {@link #allocateChunks} (the incremental strategy). Fixed at
+     * construction so the two are never mixed on the same pool.
+     */
+    public enum AllocationMode { SINGLE, CHUNKED }


Would it better to use FULL and INCREMENTAL?

junrao · 2026-07-22T23:40:09Z

+                        throw new IllegalStateException("needsNewBatch path reached without an allocated buffer stream");
+                    // Reuse the new-batch size estimate as the write-limit basis.
+                    // TODO: review when compression is supported.
+                    final NewBatchBuffer pending = newBatch;


pending => pendingNewBatch?

junrao · 2026-07-22T23:53:39Z

+     */
+    private static final class NewBatchBuffer {
+        final ChunkedByteBufferOutputStream stream;
+        final int size;


size => firstAppendSize?

junrao · 2026-07-23T23:45:55Z

+
+        // Now free chunks one at a time, allowing the multi-chunk request to accumulate.
+        p.deallocate(h1);
+        Thread.sleep(50);


Why does it need to sleep for 50ms, which is long for a unit test? Ditto below.

yeap not needed in this case really, and the other one can be better done with a waitForCondition on the available memory, done.

junrao · 2026-07-23T23:49:49Z

+            } catch (Throwable th) {
+                err.set(th);
+            }
+        }, "slow-success");


slow-success => chunked-waiter ?

lianetm · 2026-07-24T21:15:02Z

Thanks @junrao ! All comments addressed.

junrao

@lianetm : Thanks for the updated PR. A few more comments.

junrao · 2026-07-28T20:08:32Z

+            // Use the chunked path only when a batch is at least one full chunk
+            // (batch.size >= CHUNK_SIZE). Below that, a batch can't fill even one chunk, so chunking
+            // would over-reserve and the producer falls back to the full strategy instead.
+            boolean useIncremental = incremental && batchSize >= ChunkedRecordAccumulator.CHUNK_SIZE;


If we switch automatically to full because of the batch size, it might be useful to log a warning.

junrao · 2026-07-28T21:05:19Z

+        // A single call can guarantee at most `chunkSize` of space (the stream advances one chunk
+        // at a time). Callers needing more attach chunks via addBuffers first. write(byte[]) loops
+        // across chunks, so contiguous capacity isn't required.
+        ensureChunkCapacity(Math.min(remainingBytesRequired, chunkSize));


Hmm, this still seems to only work if remainingBytesRequired is 1. If it's larger than 1 and the current chunk doesn't have enough remaining bytes, we move to the next chunk. A subsequent write will write on the new chunk, the remaining bytes on the previous chunk are wasted.

junrao · 2026-07-28T21:38:00Z

+    /**
+     * Appends the record after checking there's chunk capacity for it.
+     * At this point it's expected that the allocated chunks have
+     * capacity for the record because the accumulator never routes an


At this point it's expected that the allocated chunks have capacity

Hmm, is this true? This method could return null because the batch is full, right?

junrao · 2026-07-28T22:26:38Z

-    /** Total available memory is the sum of nonPooledAvailableMemory and the number of byte buffers in free * poolableSize.  */
-    private long nonPooledAvailableMemory;
+    /** Lock held for any read or write of {@link #free}, {@link #waiters}, {@link #nonPooledAvailableMemory}, or {@link #closed}. */
+    protected final ReentrantLock lock;


Does this still need to be protected? Ditto for other fields. We can also remove the addition of BufferPool to spotbugs-exclude.xml.

junrao · 2026-07-28T22:32:01Z

+     * Wake the longest-waiting thread if any memory (pooled or non-pooled) is available.
+     * Must be called with {@link #lock} held. No-op if no waiters or no memory is free.
+     */
+    protected void signalNextWaiterIfMemoryAvailable() {


This can be private.

junrao · 2026-07-28T22:32:17Z

+     * full strategy (allocate) and the incremental strategy (ChunkedRecordAccumulator),
+     * so both update the same buffer-exhausted metrics.
+     */
+    protected void recordBufferExhausted() {


This can have package level visibility.

junrao · 2026-07-28T22:33:10Z

+     * waiter. Acquires {@link #lock} internally. Used by callers
+     * that reserve memory and then need to roll back the reservation (e.g., upon errors).
+     */
+    protected void releaseReservedBytes(long bytes) {


This can be private.

junrao · 2026-07-28T22:34:03Z

+     * buffers (if needed). Must be called with {@link #lock} held.
     */
-    private void freeUp(int size) {
+    protected void freeUp(int size) {


This can be private.

junrao · 2026-07-28T22:38:28Z

+public class ChunkedProducerBatch extends ProducerBatch {
+
+    public ChunkedProducerBatch(TopicPartition tp, MemoryRecordsBuilder recordsBuilder, long createdMs) {
+        super(tp, recordsBuilder, createdMs);


Should we validate that recordsBuilder.bufferStream() is chunked?

junrao · 2026-07-28T23:06:34Z

+                            newBatchSize, chunkedFree.poolableSize(), topic, effectivePartition, maxTimeToBlock);
+                    List<ByteBuffer> initialChunks;
+                    try {
+                        initialChunks = chunkedFree.allocateChunks(newBatchSize, maxTimeToBlock);


It seems that we can block on the bufferpool longer than maxTimeToBlock. Claude suggested the following path that doesn't exist in RecordAccumulator.

Iteration A: needsNewBatch → blocking allocateChunks(_, maxTimeToBlock) succeeds after a wait. A concurrent thread created a batch meanwhile, so appendNewBatch's in-lock tryAppend returns needsBufferExtension → line 257 deallocates the stream, continue. Iteration B: extension path does the non-blocking allocateChunks(gap, 0L) → BufferExhaustedException → closes the batch (line 168) → continue. Iteration C: batch now closed → needsNewBatch → blocks again for a fresh full maxTimeToBlock.

m1a2st · 2026-07-28T14:31:15Z

+            throw new IllegalArgumentException("totalSize must be positive: " + totalSize);
+
+        int chunkSize = poolableSize();
+        int numChunks = (int) (((long) totalSize + chunkSize - 1L) / chunkSize);


int numChunks = (totalSize - 1) / chunkSize + 1; is overflow-safe even without the cast. The existing code is correct, so feel free to ignore this comment.

lianetm added 9 commits June 22, 2026 09:52

support dyn for uncompressed

4ec8d04

tests

203a410

config & ratio-aware chunk sizing

64147f6

integration tests

8c82cd2

refactor validation & upd docs

b71d5b7

update config

bdc14e9

fix for partition change & misc

4e91499

test updates & docs

9ccd47d

fixes

860bde4

lianetm requested a review from junrao June 23, 2026 18:34

github-actions Bot added core Kafka Broker producer build Gradle build or GitHub Actions clients labels Jun 23, 2026

lianetm added 3 commits June 23, 2026 16:27

checkstyle

179e675

Merge branch 'trunk' into lm-producer-dyn-1

2bafe00

fix test

9b45abc

github-actions Bot added the tools label Jun 23, 2026

AndrewJSchofield added the ci-approved label Jun 24, 2026

junrao reviewed Jul 1, 2026

View reviewed changes

lianetm added 10 commits July 2, 2026 10:18

config internal

59af947

improvements for tryAppend

9e791c8

release unused & test

35c5940

fix to loop in ensureChunkCapacity

63de56b

fix iteration up to currentChunk

a2ee30f

remove unneeded freeup call

3c896c6

tests for concurrent reservation

20e52d3

fix for int max value cap

8d2f5e2

improve reservation when no mem available

3e1fcd8

consolidate refund of pooled and non-pooled mem

36b168d

lianetm added 4 commits July 15, 2026 14:12

refactor append result

1d8bdd3

docs

4f3543a

Merge branch 'trunk' into lm-producer-dyn-1

cb440a9

fix package name

2136f0b

chia7712 mentioned this pull request Jul 16, 2026

KAFKA-20807: Refactor ByteBufferOutputStream for extensibility (for KIP-1332) #22843

Open

chia7712 reviewed Jul 16, 2026

View reviewed changes

junrao reviewed Jul 17, 2026

View reviewed changes

lianetm added 3 commits July 20, 2026 11:27

consolidate BufferPool

9eced7d

addressing comments

7a5afd1

reuse size

68d3fee

Merge branch 'trunk' into lm-producer-dyn-1

a7aa754

junrao reviewed Jul 22, 2026

View reviewed changes

lianetm added 8 commits July 22, 2026 13:54

dedup on RecordAccumulator

d991798

pool with allocationMode

058642e

addressing comments

69828a4

more updates in tests

87074bf

assert stream closed on buffer call

8e9070b

validate chunk capacity on addBuffers & test

79caa2f

reuse estimator

b1a48d9

Merge branch 'trunk' into lm-producer-dyn-1

29bcbc4

junrao reviewed Jul 24, 2026

View reviewed changes

lianetm added 3 commits July 24, 2026 16:31

renames

e721a23

Merge branch 'trunk' into lm-producer-dyn-1

2fd0113

improve test timeouts

1de3d76

junrao reviewed Jul 28, 2026

View reviewed changes

m1a2st reviewed Jul 29, 2026

View reviewed changes

Uh oh!

Conversation

lianetm commented Jun 23, 2026 • edited by github-actions Bot Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

lianetm commented Jul 15, 2026

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

junrao left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

lianetm commented Jun 23, 2026 •

edited by github-actions Bot

Loading