@Override
    public boolean deleteBeforeSnapshot(OffsetAndEpoch snapshotId) {
        if (startOffset() > snapshotId.offset()) {
            throw new OffsetOutOfRangeException(
                String.format(
                    "New log start (%s) is less than the current log start offset (%s)",
                    snapshotId,
                    startOffset()
                )
            );
        }
        if (highWatermark.offset() < snapshotId.offset()) {
            throw new OffsetOutOfRangeException(
                String.format(
                    "New log start (%s) is greater than the high watermark (%s)",
                    snapshotId,
                    highWatermark.offset()
                )
            );
        }

        boolean updated = false;
        if (snapshots.containsKey(snapshotId)) {
            snapshots.headMap(snapshotId, false).clear();

            logger.debug("Deleting batches included in the snapshot {}", snapshotId);
            batches.removeIf(entry -> entry.lastOffset() < snapshotId.offset());

            AtomicReference<Optional<EpochStartOffset>> last = new AtomicReference<>(Optional.empty());
            epochStartOffsets.removeIf(epochStartOffset -> {
                if (epochStartOffset.startOffset <= snapshotId.offset()) {
                    last.set(Optional.of(epochStartOffset));
                    return true;
                }

                return false;
            });

            last.get().ifPresent(epochStartOffset ->
                epochStartOffsets.add(0, new EpochStartOffset(epochStartOffset.epoch, snapshotId.offset())
                )
            );

            updated = true;
        }

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