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

Code before:
        /**
         * Fails the current batch, reverts to the snapshot to the base/start offset of the
         * batch, fails all the associated events.
         */
        private void failCurrentBatch(Throwable t) {
            if (currentBatch != null) {
                coordinator.revertLastWrittenOffset(currentBatch.baseOffset);
                currentBatch.deferredEvents.complete(t);
                freeCurrentBatch();
            }
        }

        /**
         * Allocates a new batch if none already exists.
         */
        private void maybeAllocateNewBatch(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            long currentTimeMs
        ) {
            if (currentBatch == null) {
                LogConfig logConfig = partitionWriter.config(tp);
                int maxBatchSize = logConfig.maxMessageSize();
                long prevLastWrittenOffset = coordinator.lastWrittenOffset();
                ByteBuffer buffer = bufferSupplier.get(min(MIN_BUFFER_SIZE, maxBatchSize));

                MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
                    buffer,
                    RecordBatch.CURRENT_MAGIC_VALUE,
                    compression,
                    TimestampType.CREATE_TIME,
                    0L,
                    currentTimeMs,
                    producerId,
                    producerEpoch,
                    0,
                    producerId != RecordBatch.NO_PRODUCER_ID,
                    false,
                    RecordBatch.NO_PARTITION_LEADER_EPOCH,
                    maxBatchSize
                );

                Optional<TimerTask> lingerTimeoutTask = Optional.empty();
                if (appendLingerMs > 0) {
                    lingerTimeoutTask = Optional.of(new TimerTask(appendLingerMs) {
                        @Override
                        public void run() {
                            // An event to flush the batch is pushed to the front of the queue
                            // to ensure that the linger time is respected.
                            enqueueFirst(new CoordinatorInternalEvent("FlushBatch", tp, () -> {

Review comment (performance):
Using 16KB may be too small. The reallocation factor of ByteBufferOutputStream used within the batch is 1.1. The first writes will have to grow it significantly, I think. How about using something larger to start with? 

Code after:
         * Fails the current batch, reverts to the snapshot to the base/start offset of the
         * batch, fails all the associated events.
         */
        private void failCurrentBatch(Throwable t) {
            if (currentBatch != null) {
                coordinator.revertLastWrittenOffset(currentBatch.baseOffset);
                currentBatch.deferredEvents.complete(t);
                freeCurrentBatch();
            }
        }

        /**
         * Allocates a new batch if none already exists.
         */
        private void maybeAllocateNewBatch(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            long currentTimeMs
        ) {
            if (currentBatch == null) {
                LogConfig logConfig = partitionWriter.config(tp);
                int maxBatchSize = logConfig.maxMessageSize();
                long prevLastWrittenOffset = coordinator.lastWrittenOffset();
                ByteBuffer buffer = bufferSupplier.get(min(INITIAL_BUFFER_SIZE, maxBatchSize));

                MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
                    buffer,
                    RecordBatch.CURRENT_MAGIC_VALUE,
                    compression,
                    TimestampType.CREATE_TIME,
                    0L,
                    currentTimeMs,
                    producerId,
                    producerEpoch,
                    0,
                    producerId != RecordBatch.NO_PRODUCER_ID,
                    false,
                    RecordBatch.NO_PARTITION_LEADER_EPOCH,
                    maxBatchSize
                );

                Optional<TimerTask> lingerTimeoutTask = Optional.empty();
                if (appendLingerMs > 0) {
                    lingerTimeoutTask = Optional.of(new TimerTask(appendLingerMs) {
                        @Override
                        public void run() {
                            // An event to flush the batch is pushed to the front of the queue
                            // to ensure that the linger time is respected.
                            enqueueFirst(new CoordinatorInternalEvent("FlushBatch", tp, () -> {
                                if (this.isCancelled()) return;