private void maybeTruncateNonMonotonicEntries(EpochEntry newEntry) {
        List<EpochEntry> removedEpochs = removeWhileMatching(
                epochs.descendingMap().entrySet().iterator(),
                entry -> entry.epoch >= newEntry.epoch || entry.startOffset >= newEntry.startOffset);

        if (removedEpochs.size() > 1 || (!removedEpochs.isEmpty() && removedEpochs.get(0).startOffset != newEntry.startOffset)) {

            // Only log a warning if there were non-trivial removals. If the start offset of the new entry
            // matches the start offset of the removed epoch, then no data has been written and the truncation
            // is expected.
            log.warn("New epoch entry {} caused truncation of conflicting entries {}. " + "Cache now contains {} entries.", newEntry, removedEpochs, epochs.size());
        }
    }
Remove any entries which violate monotonicity prior to appending a new entry
storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java