@Override
    public Optional<SnapshotWriter<ApiMessageAndVersion>> createSnapshot(
        OffsetAndEpoch snapshotId,
        long lastContainedLogTimestamp
    ) {
        return Optional.of(
            new RecordsSnapshotWriter.Builder()
                .setLastContainedLogTimestamp(lastContainedLogTimestamp)
                .setTime(new MockTime())
                .setRawSnapshotWriter(createNewSnapshot(snapshotId))
                .build(new MetadataRecordSerde())
        );
    }
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