public long append(
        int epoch,
        List<T> records,
        OptionalLong requiredBaseOffset,
        boolean isAtomic
    ) {
        if (epoch < this.epoch) {
            throw new NotLeaderException("Append failed because the given epoch " + epoch + " is stale. " +
                    "Current leader epoch = " + this.epoch());
        } else if (epoch > this.epoch) {
            throw new IllegalArgumentException("Attempt to append from epoch " + epoch +
                " which is larger than the current epoch " + this.epoch);
        }

        ObjectSerializationCache serializationCache = new ObjectSerializationCache();

        appendLock.lock();
        try {
            long lastOffset = nextOffset + records.size() - 1;
            requiredBaseOffset.ifPresent(r -> {
                if (r != nextOffset) {
                    throw new UnexpectedBaseOffsetException("Wanted base offset " + r +
                            ", but the next offset was " + nextOffset);
                }
            });
            maybeCompleteDrain();

            BatchBuilder<T> batch = null;
            if (isAtomic) {
                batch = maybeAllocateBatch(records, serializationCache);
            }

            for (T record : records) {
                if (!isAtomic) {
                    batch = maybeAllocateBatch(Collections.singleton(record), serializationCache);
                }

                if (batch == null) {
                    throw new BufferAllocationException("Append failed because we failed to allocate memory to write the batch");
                }

                batch.appendRecord(record, serializationCache);
            }

            maybeResetLinger();

            nextOffset = lastOffset + 1;
            return lastOffset;
        } finally {
            appendLock.unlock();
        }
    }
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