@Override
    public long scheduleAppend(int epoch, List<ApiMessageAndVersion> batch) {
        if (batch.isEmpty()) {
            throw new IllegalArgumentException("Batch cannot be empty");
        }

        List<ApiMessageAndVersion> first = batch.subList(0, batch.size() / 2);
        List<ApiMessageAndVersion> second = batch.subList(batch.size() / 2, batch.size());

        assertEquals(batch.size(), first.size() + second.size());
        assertFalse(second.isEmpty());

        OptionalLong firstOffset = first
            .stream()
            .mapToLong(record -> scheduleAtomicAppend(epoch,
                    OptionalLong.empty(),
                    Collections.singletonList(record)))
            .max();

        if (firstOffset.isPresent() && resignAfterNonAtomicCommit.getAndSet(false)) {
            // Emulate losing leadership in the middle of a non-atomic append by not writing
            // the rest of the batch and instead writing a leader change message
            resign(leader.epoch());

            return firstOffset.getAsLong() + second.size();
        } else {
            return second
                .stream()
                .mapToLong(record -> scheduleAtomicAppend(epoch,
                        OptionalLong.empty(),
                        Collections.singletonList(record)))
                .max()
                .getAsLong();
        }
    }
Shutdown the log manager.

Even though the API suggests a non-blocking shutdown, this method always returns a completed
future. This means that shutdown is a blocking operation.
metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java