public long appendControlMessages(MemoryRecordsCreator valueCreator) {
        appendLock.lock();
        try {
            ByteBuffer buffer = memoryPool.tryAllocate(maxBatchSize);
            if (buffer != null) {
                try {
                    forceDrain();
                    MemoryRecords memoryRecords = valueCreator.create(
                        nextOffset,
                        epoch,
                        compression,
                        buffer
                    );

                    int numberOfRecords = validateMemoryRecordsAndReturnCount(memoryRecords);

                    completed.add(
                        new CompletedBatch<>(
                            nextOffset,
                            numberOfRecords,
                            memoryRecords,
                            memoryPool,
                            buffer
                        )
                    );
                    nextOffset += numberOfRecords;
                } catch (Exception e) {
                    // Release the buffer now since the buffer was not stored in completed for a delayed release
                    memoryPool.release(buffer);
                    throw e;
                }
            } else {
                throw new IllegalStateException("Could not allocate buffer for the control record");
            }

            return nextOffset - 1;
        } finally {
            appendLock.unlock();
        }
    }
Append a control batch from a supplied memory record.

See the {@code valueCreator} parameter description for requirements on this function.

@param valueCreator a function that uses the passed buffer to create the control
       batch that will be appended. The memory records returned must contain one
       control batch and that control batch have at least one record.
@return the last of offset of the records created
raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java