private BatchBuilder<T> maybeAllocateBatch(
        Collection<T> records,
        ObjectSerializationCache serializationCache
    ) {
        if (currentBatch == null) {
            startNewBatch();
        }

        if (currentBatch != null) {
            OptionalInt bytesNeeded = currentBatch.bytesNeeded(records, serializationCache);
            if (bytesNeeded.isPresent() && bytesNeeded.getAsInt() > maxBatchSize) {
                throw new RecordBatchTooLargeException(
                    String.format(
                        "The total record(s) size of %d exceeds the maximum allowed batch size of %d",
                        bytesNeeded.getAsInt(),
                        maxBatchSize
                    )
                );
            } else if (bytesNeeded.isPresent()) {
                completeCurrentBatch();
                startNewBatch();
            }
        }

        return currentBatch;
    }
Append to the accumulator.

@param epoch                             The leader epoch to append at.
@param records                           The records to append.
@param requiredBaseOffset                If this is non-empty, the base offset which we must use.
@param isAtomic                          True if we should append the records as a single batch.
@return                                  The end offset.

@throws NotLeaderException               Indicates that an append operation cannot be completed
                                         because the provided leader epoch was too old.
@throws IllegalArgumentException         Indicates that an append operation cannot be completed
                                         because the provided leader epoch was too new.
@throws UnexpectedBaseOffsetException    Indicates that an append operation cannot
                                         be completed because it would have resulted
                                         in an unexpected base offset.
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java