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

Code before:
                    // When adaptive linger time is enabled, we avoid flushing here.
                    // Instead, we rely on the flush event enqueued at the back of the event queue.
                    (appendLingerMs.isPresent() && (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs.getAsInt()) ||
                    !currentBatch.builder.hasRoomFor(0)) {
                    flushCurrentBatch();
                }
            }
        }

        private void failCurrentBatch(Throwable t) {
            failCurrentBatch(t, true);
        }

        private void failCurrentBatchWithoutRelease(Throwable t) {
            failCurrentBatch(t, false);
        }

        /**
         * 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, boolean freeCurrentBatch) {
            if (currentBatch != null) {
                coordinator.revertLastWrittenOffset(currentBatch.baseOffset);
                currentBatch.deferredEvents.complete(t);
                if (freeCurrentBatch) 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,

Review comment (performance):
I was discussing with @squah-confluent offline and we thought that caching `maxBatchSize` when the batch is allocated may be easier as it would avoid calling `partitionWriter.config(tp)` in `freeCurrentBatch`. It would also be more aligned with trunk. What do you guys think?

Code after:
            if (currentBatch != null) {
                if (currentBatch.builder.isTransactional() ||
                    // When adaptive linger time is enabled, we avoid flushing here.
                    // Instead, we rely on the flush event enqueued at the back of the event queue.
                    (appendLingerMs.isPresent() && (currentTimeMs - currentBatch.appendTimeMs) >= appendLingerMs.getAsInt()) ||
                    !currentBatch.builder.hasRoomFor(0)) {
                    flushCurrentBatch();
                }
            }
        }

        /**
         * 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,