Repository: apache/kafka
File: coordinator-common/src/main/java/org/apache/kafka/coordinator/common/runtime/CoordinatorRuntime.java (Java)

Code before:
                    log.error("Failed to unload coordinator for {} due to {}.", tp, ex.getMessage(), ex);
                }
            }
            coordinator = null;
        }

        /**
         * Frees the current batch.
         */
        private void freeCurrentBatch() {
            // Cancel the linger timeout.
            currentBatch.lingerTimeoutTask.ifPresent(TimerTask::cancel);

            // Release the buffer only if it is not larger than the maxBatchSize.
            int maxBatchSize = partitionWriter.config(tp).maxMessageSize();
            ByteBuffer builderBuffer = currentBatch.builder.buffer();
            ByteBuffer batchBuffer = currentBatch.buffer;

            boolean builderUsable = builderBuffer.capacity() <= maxBatchSize;
            boolean batchUsable = batchBuffer.capacity() <= maxBatchSize;

            if (builderUsable && (!batchUsable || builderBuffer.capacity() >= batchBuffer.capacity())) {
                bufferSupplier.release(builderBuffer);
            } else if (batchUsable) {
                bufferSupplier.release(batchBuffer);
            }

            currentBatch = null;
        }

        /**
         * Flushes the current (or pending) batch to the log. When the batch is written
         * locally, a new snapshot is created in the snapshot registry and the events
         * associated with the batch are added to the deferred event queue.
         */
        private void flushCurrentBatch() {
            if (currentBatch != null) {
                try {
                    if (currentBatch.builder.numRecords() == 0) {
                        // The only way we can get here is if append() has failed in an unexpected
                        // way and left an empty batch. Try to clean it up.
                        log.debug("Tried to flush an empty batch for {}.", tp);
                        // There should not be any deferred events attached to the batch. We fail
                        // the batch just in case. As a side effect, coordinator state is also
                        // reverted, but there should be no changes since the batch was empty.
                        failCurrentBatch(new IllegalStateException("Record batch was empty"));
                        return;
                    }

                    long flushStartMs = time.milliseconds();
                    // Write the records to the log and update the last written offset.

Review comment (performance):
Thanks for the patch. Isn't `currentBatch.builder.buffer().capacity()` always >= `currentBatch.buffer` because we only ever grow the buffer? We can simplify the code if that's the case.

Code after:
                    log.error("Failed to unload coordinator for {} due to {}.", tp, ex.getMessage(), ex);
                }
            }
            coordinator = null;
        }

        /**
         * Frees the current batch.
         */
        private void freeCurrentBatch() {
            // Cancel the linger timeout.
            currentBatch.lingerTimeoutTask.ifPresent(TimerTask::cancel);

            // Release the buffer only if it is not larger than the maxBatchSize.
            int maxBatchSize = partitionWriter.config(tp).maxMessageSize();

            if (currentBatch.builder.buffer().capacity() <= maxBatchSize) {
                bufferSupplier.release(currentBatch.builder.buffer());
            } else if (currentBatch.buffer.capacity() <= maxBatchSize) {
                bufferSupplier.release(currentBatch.buffer);
            }

            currentBatch = null;
        }

        /**
         * Flushes the current (or pending) batch to the log. When the batch is written
         * locally, a new snapshot is created in the snapshot registry and the events
         * associated with the batch are added to the deferred event queue.
         */
        private void flushCurrentBatch() {
            if (currentBatch != null) {
                try {
                    if (currentBatch.builder.numRecords() == 0) {
                        // The only way we can get here is if append() has failed in an unexpected
                        // way and left an empty batch. Try to clean it up.
                        log.debug("Tried to flush an empty batch for {}.", tp);
                        // There should not be any deferred events attached to the batch. We fail
                        // the batch just in case. As a side effect, coordinator state is also
                        // reverted, but there should be no changes since the batch was empty.
                        failCurrentBatch(new IllegalStateException("Record batch was empty"));
                        return;
                    }

                    long flushStartMs = time.milliseconds();
                    // Write the records to the log and update the last written offset.
                    long offset = partitionWriter.append(
                        tp,
                        currentBatch.verificationGuard,
                        currentBatch.builder.build()
                    );