@VisibleForTesting
  static void triggerLocalTierMigration(String tableNameWithType, TableTierReader.TableTierDetails tableTiers,
      ClusterMessagingService messagingService) {
    Map<String, Map<String, String>> currentTiers = tableTiers.getSegmentCurrentTiers();
    Map<String, String> targetTiers = tableTiers.getSegmentTargetTiers();
    LOGGER.debug("Got segment current tiers: {} and target tiers: {}", currentTiers, targetTiers);
    Map<String, Set<String>> serverToSegmentsToMigrate = new HashMap<>();
    for (Map.Entry<String, Map<String, String>> segmentTiers : currentTiers.entrySet()) {
      String segmentName = segmentTiers.getKey();
      Map<String, String> serverToCurrentTiers = segmentTiers.getValue();
      String targetTier = targetTiers.get(segmentName);
      for (Map.Entry<String, String> serverTier : serverToCurrentTiers.entrySet()) {
        String tier = serverTier.getValue();
        String server = serverTier.getKey();
        if ((tier == null && targetTier == null) || (tier != null && tier.equals(targetTier))) {
          LOGGER.debug("Segment: {} is already on the target tier: {} on server: {}", segmentName,
              TierConfigUtils.normalizeTierName(tier), server);
        } else {
          LOGGER.debug("Segment: {} needs to move from current tier: {} to target tier: {} on server: {}", segmentName,
              TierConfigUtils.normalizeTierName(tier), TierConfigUtils.normalizeTierName(targetTier), server);
          serverToSegmentsToMigrate.computeIfAbsent(server, (s) -> new HashSet<>()).add(segmentName);
        }
      }
    }
    if (serverToSegmentsToMigrate.size() > 0) {
      LOGGER.info("Notify servers: {} to move segments to new tiers locally", serverToSegmentsToMigrate.keySet());
      reloadSegmentsForLocalTierMigration(tableNameWithType, serverToSegmentsToMigrate, messagingService);
    } else {
      LOGGER.info("No server needs to move segments to new tiers locally");
    }
  }
@Test
  public void testTriggerLocalTierMigration() {
    String tableName = "table01";
    TableTierReader.TableTierDetails tableTiers = mock(TableTierReader.TableTierDetails.class);
    Map<String, Map<String, String>> currentTiers = new HashMap<>();
    // seg01 needs to move on both servers
    Map<String, String> serverTiers = new HashMap<>();
    serverTiers.put("server01", "hotTier");
    serverTiers.put("server02", "hotTier");
    currentTiers.put("seg01", serverTiers);
    // seg02 needs to move on a single server
    serverTiers = new HashMap<>();
    serverTiers.put("server01", "coldTier");
    serverTiers.put("server02", "hotTier");
    currentTiers.put("seg02", serverTiers);
    // seg03 needs no migration
    serverTiers = new HashMap<>();
    serverTiers.put("server01", "coldTier");
    serverTiers.put("server02", "coldTier");
    currentTiers.put("seg03", serverTiers);
    // seg04 needs to move on both servers
    serverTiers = new HashMap<>();
    serverTiers.put("server01", "coldTier");
    serverTiers.put("server02", "coldTier");
    currentTiers.put("seg04", serverTiers);

    Map<String, String> targetTiers = new HashMap<>();
    targetTiers.put("seg01", "coldTier");
    targetTiers.put("seg02", "coldTier");
    targetTiers.put("seg03", "coldTier");
    // Missing target tier for seg04, thus back to default tier: null.
    when(tableTiers.getSegmentCurrentTiers()).thenReturn(currentTiers);
    when(tableTiers.getSegmentTargetTiers()).thenReturn(targetTiers);
    ClusterMessagingService messagingService = mock(ClusterMessagingService.class);
    SegmentRelocator.triggerLocalTierMigration(tableName, tableTiers, messagingService);

    ArgumentCaptor<Criteria> criteriaCapture = ArgumentCaptor.forClass(Criteria.class);
    ArgumentCaptor<SegmentReloadMessage> reloadMessageCapture = ArgumentCaptor.forClass(SegmentReloadMessage.class);
    verify(messagingService, times(2)).send(criteriaCapture.capture(), reloadMessageCapture.capture(), eq(null),
        eq(-1));

    List<Criteria> criteriaList = criteriaCapture.getAllValues();
    List<SegmentReloadMessage> msgList = reloadMessageCapture.getAllValues();
    for (int i = 0; i < criteriaList.size(); i++) {
      String server = criteriaList.get(i).getInstanceName();
      List<String> segList = msgList.get(i).getSegmentList();
      if (server.equals("server01")) {
        assertEquals(segList.size(), 2);
        assertTrue(segList.containsAll(Arrays.asList("seg01", "seg04")));
      } else if (server.equals("server02")) {
        assertEquals(segList.size(), 3);
        assertTrue(segList.containsAll(Arrays.asList("seg01", "seg02", "seg04")));
      } else {
        fail("Unexpected server: " + server);
      }
    }
  }