public void truncateFromStartAsyncFlush(long startOffset) {
        lock.writeLock().lock();
        try {
            List<EpochEntry> removedEntries = truncateFromStart(epochs, startOffset);
            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 as part of deleteRecords with holding UnifiedLog#lock.
                //      - Meanwhile all produces against the partition will be blocked, which causes req-handlers to exhaust
                // - 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 truncateFromStart call on log loading procedure, so it won't be a problem
                scheduler.scheduleOnce("leader-epoch-cache-flush-" + topicPartition, this::writeIfDirExists);

                EpochEntry updatedFirstEntry = removedEntries.get(removedEntries.size() - 1);
                log.debug("Cleared entries {} and rewrote first entry {} after truncating to start offset {}, leaving {} in the cache.", removedEntries, updatedFirstEntry, startOffset, epochs.size());
            }
        } finally {
            lock.writeLock().unlock();
        }
    }
Clears old epoch entries. This method searches for the oldest epoch < offset, updates the saved epoch offset to
be offset, then clears any previous epoch entries.
<p>
This method is exclusive: so truncateFromStart(6) will retain an entry at offset 6.
<p>
Checkpoint-flushing is done asynchronously.

@param startOffset the offset to clear up to
storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java