private void maybeAllocateNewBatch(
            long producerId,
            short producerEpoch,
            VerificationGuard verificationGuard,
            long currentTimeMs
        ) {
            if (currentBatch == null) {
                LogConfig logConfig = partitionWriter.config(tp);
                byte magic = logConfig.recordVersion().value;
                int maxBatchSize = logConfig.maxMessageSize();
                long prevLastWrittenOffset = coordinator.lastWrittenOffset();
                ByteBuffer buffer = bufferSupplier.get(maxBatchSize);

                MemoryRecordsBuilder builder = new MemoryRecordsBuilder(
                    buffer,
                    magic,
                    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;
                                withActiveContextOrThrow(tp, CoordinatorContext::flushCurrentBatch);
                            }));
                        }
                    });
                    CoordinatorRuntime.this.timer.add(lingerTimeoutTask.get());
                }

                currentBatch = new CoordinatorBatch(
                    prevLastWrittenOffset,
                    currentTimeMs,
                    maxBatchSize,
                    verificationGuard,
                    buffer,
                    builder,
                    lingerTimeoutTask
                );
            }
        }
Allocates a new batch if none already exists.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/runtime/CoordinatorRuntime.java