public static TableRebalanceProgressStats.RebalanceStateStats getDifferenceBetweenTableRebalanceStates(
      Map<String, Map<String, String>> targetState, Map<String, Map<String, String>> sourceState) {

    TableRebalanceProgressStats.RebalanceStateStats rebalanceStats =
        new TableRebalanceProgressStats.RebalanceStateStats();

    for (Map.Entry<String, Map<String, String>> entry : targetState.entrySet()) {
      String segmentName = entry.getKey();
      Map<String, String> sourceInstanceStateMap = sourceState.get(segmentName);
      if (sourceInstanceStateMap == null) {
        // Skip the missing segment
        rebalanceStats._segmentsMissing++;
        rebalanceStats._segmentsToRebalance++;
        continue;
      }
      Map<String, String> targetStateInstanceStateMap = entry.getValue();
      boolean hasSegmentConverged = true;
      for (Map.Entry<String, String> instanceStateEntry : targetStateInstanceStateMap.entrySet()) {
        // Ignore OFFLINE state in target state
        String targetStateInstanceState = instanceStateEntry.getValue();
        if (targetStateInstanceState.equals(CommonConstants.Helix.StateModel.SegmentStateModel.OFFLINE)) {
          continue;
        }
        // Check whether the instance state in source matches the target
        String instanceName = instanceStateEntry.getKey();
        String sourceInstanceState = sourceInstanceStateMap.get(instanceName);
        if (!targetStateInstanceState.equals(sourceInstanceState)) {
          rebalanceStats._replicasToRebalance++;
          hasSegmentConverged = false;
        }
      }
      if (!hasSegmentConverged) {
        rebalanceStats._segmentsToRebalance++;
      }
    }
    int totalSegments = targetState.size();
    rebalanceStats._percentSegmentsToRebalance =
        (totalSegments == 0) ? 0 : ((double) rebalanceStats._segmentsToRebalance / totalSegments) * 100.0;
    return rebalanceStats;
  }
@Test
  void testDifferenceBetweenTableRebalanceStates() {
    Map<String, Map<String, String>> target = new TreeMap<>();
    target.put("segment1",
        SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host1", "host2", "host3"), ONLINE));
    target.put("segment2",
        SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host2", "host3", "host4"), ONLINE));

    // Stats when there's nothing to rebalance
    TableRebalanceProgressStats.RebalanceStateStats stats =
        ZkBasedTableRebalanceObserver.getDifferenceBetweenTableRebalanceStates(target, target);
    assertEquals(stats._segmentsToRebalance, 0);
    assertEquals(stats._segmentsMissing, 0);
    assertEquals(stats._percentSegmentsToRebalance, 0.0);

    // Stats when there's something to converge
    Map<String, Map<String, String>> current = new TreeMap<>();
    current.put("segment1", SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host1"), ONLINE));
    current.put("segment2", SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host2"), ONLINE));

    stats = ZkBasedTableRebalanceObserver.getDifferenceBetweenTableRebalanceStates(target, current);
    assertEquals(stats._segmentsToRebalance, 2);
    assertEquals(stats._percentSegmentsToRebalance, 100.0);
    assertEquals(stats._replicasToRebalance, 4);

    // Stats when there are errors
    current = new TreeMap<>();
    current.put("segment1", SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host1"), ERROR));

    stats = ZkBasedTableRebalanceObserver.getDifferenceBetweenTableRebalanceStates(target, current);
    assertEquals(stats._segmentsToRebalance, 2);
    assertEquals(stats._segmentsMissing, 1);
    assertEquals(stats._replicasToRebalance, 3);

    // Stats when partially converged
    current = new TreeMap<>();
    current.put("segment1",
        SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host1", "host2", "host3"), ONLINE));
    current.put("segment2", SegmentAssignmentUtils.getInstanceStateMap(Arrays.asList("host2", "host3"), ONLINE));

    stats = ZkBasedTableRebalanceObserver.getDifferenceBetweenTableRebalanceStates(target, current);
    assertEquals(stats._percentSegmentsToRebalance, 50.0);
  }