@Override
    public Optional<RawSnapshotWriter> createNewSnapshot(OffsetAndEpoch snapshotId) {
        if (snapshotId.offset() < startOffset()) {
            logger.info(
                "Cannot create a snapshot with an id ({}) less than the log start offset ({})",
                snapshotId,
                startOffset()
            );

            return Optional.empty();
        }

        long highWatermarkOffset = highWatermark().offset();
        if (snapshotId.offset() > highWatermarkOffset) {
            throw new IllegalArgumentException(
                String.format(
                    "Cannot create a snapshot with an id (%s) greater than the high-watermark (%s)",
                    snapshotId,
                    highWatermarkOffset
                )
            );
        }

        ValidOffsetAndEpoch validOffsetAndEpoch = validateOffsetAndEpoch(snapshotId.offset(), snapshotId.epoch());
        if (validOffsetAndEpoch.kind() != ValidOffsetAndEpoch.Kind.VALID) {
            throw new IllegalArgumentException(
                String.format(
                    "Snapshot id (%s) is not valid according to the log: %s",
                    snapshotId,
                    validOffsetAndEpoch
                )
            );
        }

        return createNewSnapshotUnchecked(snapshotId);
    }
Reopening the log causes all unflushed data to be lost.
raft/src/test/java/org/apache/kafka/raft/MockLog.java