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

Code before:
                partitionWriter.deregisterListener(tp, highWatermarklistener);
                highWatermarklistener = null;
            }
            timer.cancelAll();
            executor.cancelAll();
            deferredEventQueue.failAll(Errors.NOT_COORDINATOR.exception());
            failCurrentBatch(Errors.NOT_COORDINATOR.exception());
            if (coordinator != null) {
                try {
                    coordinator.onUnloaded();
                } catch (Throwable ex) {
                    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 max buffer size.
            int maxBufferSize = appendMaxBufferSizeSupplier.get();

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

            currentBatch = null;
        }

        /**
         * Adds a flush event to the end of the event queue, after any existing writes in the queue.
         *
         * @param expectedBatchEpoch The epoch of the batch to flush.
         */
        private void enqueueAdaptiveFlush(int expectedBatchEpoch) {
            enqueueLast(new CoordinatorInternalEvent("FlushBatch", tp, () -> {
                withActiveContextOrThrow(tp, context -> {
                    // The batch could have already been flushed because it reached the maximum
                    // batch size or a transactional write came in. When this happens, we want
                    // to avoid flushing the next batch early.
                    if (context.currentBatch != null && context.batchEpoch == expectedBatchEpoch) {
                        context.flushCurrentBatch();

Review comment (performance):
Should `maybeAllocateNewBatch` also adopt `appendMaxBufferSizeSupplier` instead of message size?

Code after:
            }
            timer.cancelAll();
            executor.cancelAll();
            deferredEventQueue.failAll(Errors.NOT_COORDINATOR.exception());
            failCurrentBatch(Errors.NOT_COORDINATOR.exception());
            if (coordinator != null) {
                try {
                    coordinator.onUnloaded();
                } catch (Throwable ex) {
                    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 cachedBufferMaxBytes.
            int cachedBufferMaxBytes = cachedBufferMaxBytesSupplier.get();

            if (currentBatch.builder.buffer().capacity() <= cachedBufferMaxBytes) {
                bufferSupplier.release(currentBatch.builder.buffer());
                cachedBufferSize.set(currentBatch.builder.buffer().capacity());
            } else if (currentBatch.buffer.capacity() <= cachedBufferMaxBytes) {
                bufferSupplier.release(currentBatch.buffer);
                cachedBufferSize.set(currentBatch.buffer.capacity());
                // If the builder expands the buffer beyond the cachedBufferMaxBytes, that should also increase the discard counter.
                runtimeMetrics.recordBufferCacheDiscarded();
            } else {
                runtimeMetrics.recordBufferCacheDiscarded();
                cachedBufferSize.set(0L);
            }

            currentBatch = null;
        }

        /**
         * Adds a flush event to the end of the event queue, after any existing writes in the queue.
         *
         * @param expectedBatchEpoch The epoch of the batch to flush.
         */
        private void enqueueAdaptiveFlush(int expectedBatchEpoch) {
            enqueueLast(new CoordinatorInternalEvent("FlushBatch", tp, () -> {
                withActiveContextOrThrow(tp, context -> {
                    // The batch could have already been flushed because it reached the maximum
                    // batch size or a transactional write came in. When this happens, we want