static boolean isRemoteSegmentWithinLeaderEpochs(RemoteLogSegmentMetadata segmentMetadata,
                                                     long logEndOffset,
                                                     NavigableMap<Integer, Long> leaderEpochs) {
        long segmentEndOffset = segmentMetadata.endOffset();
        // Filter epochs that does not have any messages/records associated with them.
        NavigableMap<Integer, Long> segmentLeaderEpochs = buildFilteredLeaderEpochMap(segmentMetadata.segmentLeaderEpochs());
        // Check for out of bound epochs between segment epochs and current leader epochs.
        Integer segmentLastEpoch = segmentLeaderEpochs.lastKey();
        if (segmentLastEpoch < leaderEpochs.firstKey() || segmentLastEpoch > leaderEpochs.lastKey()) {
            LOGGER.debug("Segment {} is not within the partition leader epoch lineage. " +
                            "Remote segment epochs: {} and partition leader epochs: {}",
                    segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs, leaderEpochs);
            return false;
        }
        // There can be overlapping remote log segments in the remote storage. (eg)
        // leader-epoch-file-cache: {(5, 10), (7, 15), (9, 100)}
        // segment1: offset-range = 5-50, Broker = 0, epochs = {(5, 10), (7, 15)}
        // segment2: offset-range = 14-150, Broker = 1, epochs = {(5, 14), (7, 15), (9, 100)}, after leader-election.
        // When the segment1 gets deleted, then the log-start-offset = 51 and leader-epoch-file-cache gets updated to: {(7, 51), (9, 100)}.
        // While validating the segment2, we should ensure the overlapping remote log segments case.
        Integer segmentFirstEpoch = segmentLeaderEpochs.ceilingKey(leaderEpochs.firstKey());
        if (segmentFirstEpoch == null) {
            LOGGER.debug("Segment {} is not within the partition leader epoch lineage. " +
                            "Remote segment epochs: {} and partition leader epochs: {}",
                    segmentMetadata.remoteLogSegmentId(), segmentLeaderEpochs, leaderEpochs);
            return false;
        }
        for (Map.Entry<Integer, Long> entry : segmentLeaderEpochs.entrySet()) {
            int epoch = entry.getKey();
            long offset = entry.getValue();
            if (epoch < segmentFirstEpoch) {
                continue;
            }
            // If segment's epoch does not exist in the leader epoch lineage then it is not a valid segment.
            if (!leaderEpochs.containsKey(epoch)) {
                LOGGER.debug("Segment {} epoch {} is not within the leader epoch lineage. " +
                                "Remote segment epochs: {} and partition leader epochs: {}",
                        segmentMetadata.remoteLogSegmentId(), epoch, segmentLeaderEpochs, leaderEpochs);
                return false;
            }
            // Two cases:
            // case-1: When the segment-first-epoch equals to the first-epoch in the leader-epoch-lineage, then the
            // offset value can lie anywhere between 0 to (next-epoch-start-offset - 1) is valid.
            // case-2: When the segment-first-epoch is not equal to the first-epoch in the leader-epoch-lineage, then
            // the offset value should be between (current-epoch-start-offset) to (next-epoch-start-offset - 1).
            if (epoch == segmentFirstEpoch && leaderEpochs.lowerKey(epoch) != null && offset < leaderEpochs.get(epoch)) {
                LOGGER.debug("Segment {} first-valid epoch {} offset is less than first leader epoch offset {}." +
                                "Remote segment epochs: {} and partition leader epochs: {}",
                        segmentMetadata.remoteLogSegmentId(), epoch, leaderEpochs.get(epoch),
                        segmentLeaderEpochs, leaderEpochs);
                return false;
            }
            // Segment's end offset should be less than or equal to the respective leader epoch's offset.
            if (epoch == segmentLastEpoch) {
                Map.Entry<Integer, Long> nextEntry = leaderEpochs.higherEntry(epoch);
                if (nextEntry != null && segmentEndOffset > nextEntry.getValue() - 1) {
                    LOGGER.debug("Segment {} end offset {} is more than leader epoch offset {}." +
                                    "Remote segment epochs: {} and partition leader epochs: {}",
                            segmentMetadata.remoteLogSegmentId(), segmentEndOffset, nextEntry.getValue() - 1,
                            segmentLeaderEpochs, leaderEpochs);
                    return false;
                }
            }
            // Next segment epoch entry and next leader epoch entry should be same to ensure that the segment's epoch
            // is within the leader epoch lineage.
            if (epoch != segmentLastEpoch && !leaderEpochs.higherEntry(epoch).equals(segmentLeaderEpochs.higherEntry(epoch))) {
                LOGGER.debug("Segment {} epoch {} is not within the leader epoch lineage. " +
                                "Remote segment epochs: {} and partition leader epochs: {}",
                        segmentMetadata.remoteLogSegmentId(), epoch, segmentLeaderEpochs, leaderEpochs);
                return false;
            }
        }
        // segment end offset should be with in the log end offset.
        if (segmentEndOffset >= logEndOffset) {
            LOGGER.debug("Segment {} end offset {} is more than log end offset {}.",
                    segmentMetadata.remoteLogSegmentId(), segmentEndOffset, logEndOffset);
            return false;
        }
        return true;
    }
@Test
    public void testRemoteSegmentWithinLeaderEpochsForOverlappingSegments() {
        NavigableMap<Integer, Long> leaderEpochCache = new TreeMap<>();
        leaderEpochCache.put(7, 51L);
        leaderEpochCache.put(9, 100L);

        TreeMap<Integer, Long> segment1Epochs = new TreeMap<>();
        segment1Epochs.put(5, 14L);
        segment1Epochs.put(7, 15L);
        segment1Epochs.put(9, 100L);
        RemoteLogSegmentMetadata segment1 = createRemoteLogSegmentMetadata(14, 150, segment1Epochs);
        assertTrue(isRemoteSegmentWithinLeaderEpochs(segment1, 210, leaderEpochCache));

        // segment2Epochs are not within the leaderEpochCache
        TreeMap<Integer, Long> segment2Epochs = new TreeMap<>();
        segment2Epochs.put(2, 5L);
        segment2Epochs.put(3, 6L);
        RemoteLogSegmentMetadata segment2 = createRemoteLogSegmentMetadata(2, 7, segment2Epochs);
        assertFalse(isRemoteSegmentWithinLeaderEpochs(segment2, 210, leaderEpochCache));

        // segment3Epochs are not within the leaderEpochCache
        TreeMap<Integer, Long> segment3Epochs = new TreeMap<>();
        segment3Epochs.put(7, 15L);
        segment3Epochs.put(9, 100L);
        segment3Epochs.put(10, 200L);
        RemoteLogSegmentMetadata segment3 = createRemoteLogSegmentMetadata(15, 250, segment3Epochs);
        assertFalse(isRemoteSegmentWithinLeaderEpochs(segment3, 210, leaderEpochCache));

        // segment4Epochs are not within the leaderEpochCache
        TreeMap<Integer, Long> segment4Epochs = new TreeMap<>();
        segment4Epochs.put(8, 75L);
        RemoteLogSegmentMetadata segment4 = createRemoteLogSegmentMetadata(75, 100, segment4Epochs);
        assertFalse(isRemoteSegmentWithinLeaderEpochs(segment4, 210, leaderEpochCache));

        // segment5Epochs does not match with the leaderEpochCache
        TreeMap<Integer, Long> segment5Epochs = new TreeMap<>();
        segment5Epochs.put(7, 15L);
        segment5Epochs.put(9, 101L);
        RemoteLogSegmentMetadata segment5 = createRemoteLogSegmentMetadata(15, 150, segment5Epochs);
        assertFalse(isRemoteSegmentWithinLeaderEpochs(segment5, 210, leaderEpochCache));

        // segment6Epochs does not match with the leaderEpochCache
        TreeMap<Integer, Long> segment6Epochs = new TreeMap<>();
        segment6Epochs.put(9, 99L);
        RemoteLogSegmentMetadata segment6 = createRemoteLogSegmentMetadata(99, 150, segment6Epochs);
        assertFalse(isRemoteSegmentWithinLeaderEpochs(segment6, 210, leaderEpochCache));
    }