OffsetAndEpoch findHighestRemoteOffset(TopicIdPartition topicIdPartition, UnifiedLog log) throws RemoteStorageException {
        OffsetAndEpoch offsetAndEpoch = null;
        Option<LeaderEpochFileCache> leaderEpochCacheOpt = log.leaderEpochCache();
        if (leaderEpochCacheOpt.isDefined()) {
            LeaderEpochFileCache cache = leaderEpochCacheOpt.get();
            Optional<EpochEntry> maybeEpochEntry = cache.latestEntry();
            while (offsetAndEpoch == null && maybeEpochEntry.isPresent()) {
                int epoch = maybeEpochEntry.get().epoch;
                Optional<Long> highestRemoteOffsetOpt =
                        remoteLogMetadataManager.highestOffsetForEpoch(topicIdPartition, epoch);
                if (highestRemoteOffsetOpt.isPresent()) {
                    Map.Entry<Integer, Long> entry = cache.endOffsetFor(epoch, log.logEndOffset());
                    int requestedEpoch = entry.getKey();
                    long endOffset = entry.getValue();
                    long highestRemoteOffset = highestRemoteOffsetOpt.get();
                    if (endOffset <= highestRemoteOffset) {
                        LOGGER.info("The end-offset for epoch {}: ({}, {}) is less than or equal to the " +
                                "highest-remote-offset: {} for partition: {}", epoch, requestedEpoch, endOffset,
                                highestRemoteOffset, topicIdPartition);
                        offsetAndEpoch = new OffsetAndEpoch(endOffset - 1, requestedEpoch);
                    } else {
                        offsetAndEpoch = new OffsetAndEpoch(highestRemoteOffset, epoch);
                    }
                }
                maybeEpochEntry = cache.previousEntry(epoch);
            }
        }
        if (offsetAndEpoch == null) {
            offsetAndEpoch = new OffsetAndEpoch(-1L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
        }
        return offsetAndEpoch;
    }
Returns a map containing the epoch vs start-offset for the given leader epoch map by filtering the epochs that
does not contain any messages/records associated with them.
For ex:
<pre>
{@code
 <epoch - start offset>
 0 - 0
 1 - 10
 2 - 20
 3 - 30
 4 - 40
 5 - 60  // epoch 5 does not have records or messages associated with it
 6 - 60
 7 - 70
}
</pre>
When the above leaderEpochMap is passed to this method, it returns the following map:
<pre>
{@code
 <epoch - start offset>
 0 - 0
 1 - 10
 2 - 20
 3 - 30
 4 - 40
 6 - 60
 7 - 70
}
</pre>
@param leaderEpochs The leader epoch map to be refined.
core/src/main/java/kafka/log/remote/RemoteLogManager.java