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();
        }
    }
@Test
    public void testInvalidControlRecordOffset() {
        int leaderEpoch = 17;
        long baseOffset = 157;
        int lingerMs = 50;
        int maxBatchSize = 512;

        ByteBuffer buffer = ByteBuffer.allocate(maxBatchSize);
        Mockito.when(memoryPool.tryAllocate(maxBatchSize))
                .thenReturn(buffer);

        BatchAccumulator.MemoryRecordsCreator creator = (offset, epoch, compression, buf) -> {
            long now = 1234;
            try (MemoryRecordsBuilder builder = controlRecordsBuilder(
                    offset + 1,
                    epoch,
                    compression,
                    now,
                    buf
                )
            ) {
                builder.appendSnapshotHeaderMessage(
                    now,
                    new SnapshotHeaderRecord()
                        .setVersion(ControlRecordUtils.SNAPSHOT_HEADER_CURRENT_VERSION)
                        .setLastContainedLogTimestamp(now)
                );

                return builder.build();
            }
        };

        try (BatchAccumulator<String> acc = buildAccumulator(
                leaderEpoch,
                baseOffset,
                lingerMs,
                maxBatchSize
            )
        ) {
            assertThrows(IllegalArgumentException.class, () -> acc.appendControlMessages(creator));
        }
    }