Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/producer/internals/RecordAccumulator.java (Java)

Code before:
    }

    BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
        return new BuiltInPartitioner(logContext, topic, stickyBatchSize);
    }

    /**
     * Complete and deallocate the record batch
     */
    public void completeAndDeallocateBatch(ProducerBatch batch) {
        completeBatch(batch);
        deallocate(batch);
    }

    /**
     * Only perform deallocation (and not removal from the incomplete set)
     */
    public void deallocate(ProducerBatch batch) {
        // Only deallocate the batch if it is not a split batch because split batch are allocated outside the
        // buffer pool.
        if (!batch.isSplitBatch()) {
            if (batch.isBufferDeallocated()) {
                log.warn("Skipping deallocating a batch that has already been deallocated. Batch is {}, created time is {}", batch, batch.createdMs);
            } else {
                if (batch.isInflight()) {
                    throw new IllegalStateException("Attempting to deallocate a batch that is not inflight. Batch is " + batch);
                }
                batch.markBufferDeallocated();
                free.deallocate(batch.buffer(), batch.initialCapacity());
            }
        }
    }

    /**
     * Remove from the incomplete list but do not free memory yet
     */
    public void completeBatch(ProducerBatch batch) {
        incomplete.remove(batch);
    }

    /**
     * Package private for unit test. Get the buffer pool remaining size in bytes.
     */
    long bufferPoolAvailableMemory() {
        return free.availableMemory();
    }

    /**
     * Are there any threads currently waiting on a flush?
     *
     * package private for test

Review comment (performance):
Since the Sender thread just logs an exception and continues, we could be silently leaking memory if this happens. An alternative is to log a warn and create a fresh buffer of batch.initialCapacity() size to deallocate. It would be useful to include the stacktrace in the log message. 

> that is not inflight

that is inflight

Code after:
    }

    BuiltInPartitioner createBuiltInPartitioner(LogContext logContext, String topic, int stickyBatchSize) {
        return new BuiltInPartitioner(logContext, topic, stickyBatchSize);
    }

    /**
     * Complete and deallocate the record batch
     */
    public void completeAndDeallocateBatch(ProducerBatch batch) {
        completeBatch(batch);
        deallocate(batch);
    }

    /**
     * Only perform deallocation (and not removal from the incomplete set)
     */
    public void deallocate(ProducerBatch batch) {
        // Only deallocate the batch if it is not a split batch because split batch are allocated outside the
        // buffer pool.
        if (!batch.isSplitBatch()) {
            if (batch.isBufferDeallocated()) {
                log.warn("Skipping deallocating a batch that has already been deallocated. Batch is {}, created time is {}", batch, batch.createdMs);
            } else {
                batch.markBufferDeallocated();
                if (batch.isInflight()) {
                    // Create a fresh ByteBuffer to give to BufferPool to reuse since we can't safely call deallocate with the ProduceBatch's buffer
                    free.deallocate(ByteBuffer.allocate(batch.initialCapacity()));
                    throw new IllegalStateException("Attempting to deallocate a batch that is inflight. Batch is " + batch);
                }
                free.deallocate(batch.buffer(), batch.initialCapacity());
            }
        }
    }

    /**
     * Remove from the incomplete list but do not free memory yet
     */
    public void completeBatch(ProducerBatch batch) {
        incomplete.remove(batch);
    }

    /**
     * Package private for unit test. Get the buffer pool remaining size in bytes.
     */
    long bufferPoolAvailableMemory() {
        return free.availableMemory();
    }

    /**
     * Are there any threads currently waiting on a flush?