@VisibleForTesting
  MissingSegmentInfo findMissingSegments(Map<String, Map<String, String>> idealStateMap, Instant now) {
    // create the maps
    Map<Integer, LLCSegmentName> partitionGroupIdToLatestConsumingSegmentMap = new HashMap<>();
    Map<Integer, LLCSegmentName> partitionGroupIdToLatestCompletedSegmentMap = new HashMap<>();
    idealStateMap.forEach((segmentName, instanceToStatusMap) -> {
      LLCSegmentName llcSegmentName = LLCSegmentName.of(segmentName);
      if (llcSegmentName != null) { // Skip the uploaded realtime segments that don't conform to llc naming
        if (instanceToStatusMap.containsValue(SegmentStateModel.CONSUMING)) {
          updateMap(partitionGroupIdToLatestConsumingSegmentMap, llcSegmentName);
        } else if (instanceToStatusMap.containsValue(SegmentStateModel.ONLINE)) {
          updateMap(partitionGroupIdToLatestCompletedSegmentMap, llcSegmentName);
        }
      }
    });

    MissingSegmentInfo missingSegmentInfo = new MissingSegmentInfo();
    if (!_partitionGroupIdToLargestStreamOffsetMap.isEmpty()) {
      _partitionGroupIdToLargestStreamOffsetMap.forEach((partitionGroupId, largestStreamOffset) -> {
        if (!partitionGroupIdToLatestConsumingSegmentMap.containsKey(partitionGroupId)) {
          LLCSegmentName latestCompletedSegment = partitionGroupIdToLatestCompletedSegmentMap.get(partitionGroupId);
          if (latestCompletedSegment == null) {
            // There's no consuming or completed segment for this partition group. Possibilities:
            //   1) it's a new partition group that has not yet been detected
            //   2) the first consuming segment has been deleted from ideal state manually
            missingSegmentInfo._newPartitionGroupCount++;
            missingSegmentInfo._totalCount++;
          } else {
            // Completed segment is available, but there's no consuming segment.
            // Note that there is no problem in case the partition group has reached its end of life.
            SegmentZKMetadata segmentZKMetadata = _segmentMetadataFetcher
                .fetchSegmentZkMetadata(_realtimeTableName, latestCompletedSegment.getSegmentName());
            StreamPartitionMsgOffset completedSegmentEndOffset =
                _streamPartitionMsgOffsetFactory.create(segmentZKMetadata.getEndOffset());
            if (completedSegmentEndOffset.compareTo(largestStreamOffset) < 0) {
              // there are unconsumed messages available on the stream
              missingSegmentInfo._totalCount++;
              updateMaxDurationInfo(missingSegmentInfo, partitionGroupId, segmentZKMetadata.getCreationTime(), now);
            }
          }
        }
      });
    } else {
      partitionGroupIdToLatestCompletedSegmentMap.forEach((partitionGroupId, latestCompletedSegment) -> {
        if (!partitionGroupIdToLatestConsumingSegmentMap.containsKey(partitionGroupId)) {
          missingSegmentInfo._totalCount++;
          long segmentCompletionTimeMillis = _segmentMetadataFetcher
              .fetchSegmentCompletionTime(_realtimeTableName, latestCompletedSegment.getSegmentName());
          updateMaxDurationInfo(missingSegmentInfo, partitionGroupId, segmentCompletionTimeMillis, now);
        }
      });
    }
    return missingSegmentInfo;
  }
@Test
  public void noMissingConsumingSegmentsScenario3() {
    // scenario 3: no missing segments and there's no exception in connecting to stream
    // two partitions have reached end of life

    Map<String, Map<String, String>> idealStateMap = new HashMap<>();
    // partition 0
    idealStateMap.put("tableA__0__0__20220601T0900Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__0__1__20220601T1200Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__0__2__20220601T1500Z", ImmutableMap.of("ServerX", "CONSUMING", "ServerY", "CONSUMING"));
    // partition 1 (has reached end of life)
    idealStateMap.put("tableA__1__0__20220601T0900Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__1__1__20220601T1200Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    // partition 2
    idealStateMap.put("tableA__2__0__20220601T0900Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__2__1__20220601T1200Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__2__2__20220601T1500Z", ImmutableMap.of("ServerX", "CONSUMING", "ServerY", "CONSUMING"));
    // partition 3 (has reached end of life)
    idealStateMap.put("tableA__3__0__20220601T0900Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));
    idealStateMap.put("tableA__3__1__20220601T1200Z", ImmutableMap.of("ServerX", "ONLINE", "ServerY", "ONLINE"));

    Map<Integer, StreamPartitionMsgOffset> partitionGroupIdToLargestStreamOffsetMap = ImmutableMap.of(
        0, new LongMsgOffset(1000),
        1, new LongMsgOffset(701),
        2, new LongMsgOffset(1002),
        3, new LongMsgOffset(703)
    );

    // setup segment metadata fetcher
    SegmentZKMetadata m1 = mock(SegmentZKMetadata.class);
    when(m1.getEndOffset()).thenReturn("701");
    SegmentZKMetadata m3 = mock(SegmentZKMetadata.class);
    when(m3.getEndOffset()).thenReturn("703");
    MissingConsumingSegmentFinder.SegmentMetadataFetcher metadataFetcher =
        mock(MissingConsumingSegmentFinder.SegmentMetadataFetcher.class);
    when(metadataFetcher.fetchSegmentZkMetadata("tableA", "tableA__1__1__20220601T1200Z")).thenReturn(m1);
    when(metadataFetcher.fetchSegmentZkMetadata("tableA", "tableA__3__1__20220601T1200Z")).thenReturn(m3);

    Instant now = Instant.parse("2022-06-01T18:00:00.00Z");
    MissingConsumingSegmentFinder finder =
        new MissingConsumingSegmentFinder("tableA", metadataFetcher, partitionGroupIdToLargestStreamOffsetMap,
            _offsetFactory);
    MissingConsumingSegmentFinder.MissingSegmentInfo info = finder.findMissingSegments(idealStateMap, now);
    assertEquals(info._totalCount, 0);
    assertEquals(info._newPartitionGroupCount, 0);
    assertEquals(info._maxDurationInMinutes, 0);
  }