public void reopen() {
        batches.removeIf(batch -> batch.firstOffset() >= firstUnflushedOffset);
        epochStartOffsets.removeIf(epochStartOffset -> epochStartOffset.startOffset >= firstUnflushedOffset);
        highWatermark = new LogOffsetMetadata(0L, Optional.empty());
    }
Reopening the log causes all unflushed data to be lost.
raft/src/test/java/org/apache/kafka/raft/MockLog.java