public Map.Entry<Integer, Long> endOffsetFor(int requestedEpoch, long logEndOffset) {
        lock.readLock().lock();
        try {
            Map.Entry<Integer, Long> epochAndOffset = null;
            if (requestedEpoch == UNDEFINED_EPOCH) {
                // This may happen if a bootstrapping follower sends a request with undefined epoch or
                // a follower is on the older message format where leader epochs are not recorded
                epochAndOffset = new AbstractMap.SimpleImmutableEntry<>(UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET);
            } else if (latestEpoch().isPresent() && latestEpoch().getAsInt() == requestedEpoch) {
                // For the leader, the latest epoch is always the current leader epoch that is still being written to.
                // Followers should not have any reason to query for the end offset of the current epoch, but a consumer
                // might if it is verifying its committed offset following a group rebalance. In this case, we return
                // the current log end offset which makes the truncation check work as expected.
                epochAndOffset = new AbstractMap.SimpleImmutableEntry<>(requestedEpoch, logEndOffset);
            } else {
                Map.Entry<Integer, EpochEntry> higherEntry = epochs.higherEntry(requestedEpoch);
                if (higherEntry == null) {
                    // The requested epoch is larger than any known epoch. This case should never be hit because
                    // the latest cached epoch is always the largest.
                    epochAndOffset = new AbstractMap.SimpleImmutableEntry<>(UNDEFINED_EPOCH, UNDEFINED_EPOCH_OFFSET);
                } else {
                    Map.Entry<Integer, EpochEntry> floorEntry = epochs.floorEntry(requestedEpoch);
                    if (floorEntry == null) {
                        // The requested epoch is smaller than any known epoch, so we return the start offset of the first
                        // known epoch which is larger than it. This may be inaccurate as there could have been
                        // epochs in between, but the point is that the data has already been removed from the log
                        // and we want to ensure that the follower can replicate correctly beginning from the leader's
                        // start offset.
                        epochAndOffset = new AbstractMap.SimpleImmutableEntry<>(requestedEpoch, higherEntry.getValue().startOffset);
                    } else {
                        // We have at least one previous epoch and one subsequent epoch. The result is the first
                        // prior epoch and the starting offset of the first subsequent epoch.
                        epochAndOffset = new AbstractMap.SimpleImmutableEntry<>(floorEntry.getValue().epoch, higherEntry.getValue().startOffset);
                    }
                }
            }

            if (log.isTraceEnabled())
                log.trace("Processed end offset request for epoch {} and returning epoch {} with end offset {} from epoch cache of size {}}", requestedEpoch, epochAndOffset.getKey(), epochAndOffset.getValue(), epochs.size());

            return epochAndOffset;
        } finally {
            lock.readLock().unlock();
        }
    }
Returns the Leader Epoch and the End Offset for a requested Leader Epoch.
<p>
The Leader Epoch returned is the largest epoch less than or equal to the requested Leader
Epoch. The End Offset is the end offset of this epoch, which is defined as the start offset
of the first Leader Epoch larger than the Leader Epoch requested, or else the Log End
Offset if the latest epoch was requested.
<p>
During the upgrade phase, where there are existing messages may not have a leader epoch,
if requestedEpoch is < the first epoch cached, UNDEFINED_EPOCH_OFFSET will be returned
so that the follower falls back to High Water Mark.

@param requestedEpoch requested leader epoch
@param logEndOffset   the existing Log End Offset
@return found leader epoch and end offset
storage/src/main/java/org/apache/kafka/storage/internals/epoch/LeaderEpochFileCache.java