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;
    }
@Test
    void testFindHighestRemoteOffset() throws RemoteStorageException {
        List<EpochEntry> totalEpochEntries = Arrays.asList(
                new EpochEntry(0, 0),
                new EpochEntry(1, 500)
        );
        checkpoint.write(totalEpochEntries);
        LeaderEpochFileCache cache = new LeaderEpochFileCache(tp, checkpoint, scheduler);
        when(mockLog.leaderEpochCache()).thenReturn(Option.apply(cache));
        TopicIdPartition tpId = new TopicIdPartition(Uuid.randomUuid(), tp);
        when(remoteLogMetadataManager.highestOffsetForEpoch(eq(tpId), anyInt())).thenAnswer(ans -> {
            Integer epoch = ans.getArgument(1, Integer.class);
            if (epoch == 0) {
                return Optional.of(200L);
            } else {
                return Optional.empty();
            }
        });
        OffsetAndEpoch offsetAndEpoch = remoteLogManager.findHighestRemoteOffset(tpId, mockLog);
        assertEquals(new OffsetAndEpoch(200L, 0), offsetAndEpoch);
    }