public OptionalInt epochForOffset(long offset) {
        lock.readLock().lock();
        try {
            OptionalInt previousEpoch = OptionalInt.empty();
            for (EpochEntry epochEntry : epochs.values()) {
                int epoch = epochEntry.epoch;
                long startOffset = epochEntry.startOffset;

                // Found the exact offset, return the respective epoch.
                if (startOffset == offset) return OptionalInt.of(epoch);

                // Return the previous found epoch as this epoch's start-offset is more than the target offset.
                if (startOffset > offset) return previousEpoch;

                previousEpoch = OptionalInt.of(epoch);
            }

            return previousEpoch;
        } finally {
            lock.readLock().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