@Override
    public void initializeLeaderEpoch(int epoch) {
        long startOffset = endOffset().offset();
        epochStartOffsets.removeIf(epochStartOffset ->
            epochStartOffset.startOffset >= startOffset || epochStartOffset.epoch >= epoch);
        epochStartOffsets.add(new EpochStartOffset(epoch, startOffset));
    }
Reopening the log causes all unflushed data to be lost.
raft/src/test/java/org/apache/kafka/raft/MockLog.java