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

Code before:
    /**
     * Deallocate the record batch
     */
    public void deallocate(ProducerBatch batch) {
        incomplete.remove(batch);
        deallocateIfNecessary(batch);
    }

    private void deallocateIfNecessary(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() && !batch.isBufferDeallocated() && batch.setBufferDeallocated(true))
            free.deallocate(batch.buffer(), batch.initialCapacity());
    }

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

    /**
     * Only perform deallocation (and not removal from incomplete set)
     */
    public void deallocateAlreadyRemovedIncomplete(ProducerBatch batch) {
        deallocateIfNecessary(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
     */
    boolean flushInProgress() {
        return flushesInProgress.get() > 0;
    }

    /**
     * Initiate the flushing of data from the accumulator...this makes all requests immediately ready
     */
    public void beginFlush() {
        this.flushesInProgress.getAndIncrement();
    }

Review comment (performance):
Perhaps it's clearer if we expose the following 3 methods.
```
completeBatch()
deallocate()
completeBatchAndDeallocate()
```

Code after:
    /**
     * 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?
     *
     * package private for test
     */
    boolean flushInProgress() {
        return flushesInProgress.get() > 0;
    }