public void truncateFromEndAsyncFlush(long endOffset) {
        lock.writeLock().lock();
        try {
            List<EpochEntry> removedEntries = truncateFromEnd(epochs, endOffset);
            if (!removedEntries.isEmpty()) {
                // We flush the change to the device in the background because:
                // - To avoid fsync latency
                //   * fsync latency could be huge on a disk glitch, which is not rare in spinning drives
                //   * This method is called by ReplicaFetcher threads, which could block replica fetching
                //     then causing ISR shrink or high produce response time degradation in remote scope on high fsync latency.
                // - We still flush the change in #assign synchronously, meaning that it's guaranteed that the checkpoint file always has no missing entries.
                //   * Even when stale epochs are restored from the checkpoint file after the unclean shutdown, it will be handled by
                //     another truncateFromEnd call on log loading procedure, so it won't be a problem
                scheduler.scheduleOnce("leader-epoch-cache-flush-" + topicPartition, this::writeIfDirExists);

                log.debug("Cleared entries {} from epoch cache after truncating to end offset {}, leaving {} entries in the cache.", removedEntries, endOffset, epochs.size());
            }
        } finally {
            lock.writeLock().unlock();
        }
    }
Removes all epoch entries from the store with start offsets greater than or equal to the passed offset.
<p>
Checkpoint-flushing is done asynchronously.
storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java