@VisibleForTesting
  static Pair<TableRebalanceContext, Long> getLatestJob(
      Map<String, Set<Pair<TableRebalanceContext, Long>>> candidateJobs) {
    Pair<TableRebalanceContext, Long> candidateJobRun = null;
    for (Map.Entry<String, Set<Pair<TableRebalanceContext, Long>>> entry : candidateJobs.entrySet()) {
      // The job configs from all retry jobs are same, as the same set of job configs is used to do retry.
      // The job metadata kept in ZK is cleaned by submission time order gradually, so we can't compare Set.size()
      // against maxAttempts, but check retryNum of each run to see if retries have exceeded limit.
      Set<Pair<TableRebalanceContext, Long>> jobRuns = entry.getValue();
      int maxAttempts = jobRuns.iterator().next().getLeft().getConfig().getMaxAttempts();
      Pair<TableRebalanceContext, Long> latestJobRun = null;
      for (Pair<TableRebalanceContext, Long> jobRun : jobRuns) {
        if (jobRun.getLeft().getAttemptId() >= maxAttempts) {
          latestJobRun = null;
          break;
        }
        if (latestJobRun == null || latestJobRun.getRight() < jobRun.getRight()) {
          latestJobRun = jobRun;
        }
      }
      if (latestJobRun == null) {
        LOGGER.info("Rebalance job: {} had exceeded maxAttempts: {}. Skip retry", entry.getKey(), maxAttempts);
        continue;
      }
      if (candidateJobRun == null || candidateJobRun.getRight() < latestJobRun.getRight()) {
        candidateJobRun = latestJobRun;
      }
    }
    return candidateJobRun;
  }
@Test
  public void testGetLatestJob() {
    Map<String, Set<Pair<TableRebalanceContext, Long>>> jobs = new HashMap<>();
    // The most recent job run is job1_3, and within 3 maxAttempts.
    jobs.put("job1",
        ImmutableSet.of(Pair.of(createDummyJobCtx("job1", 1), 10L), Pair.of(createDummyJobCtx("job1", 2), 20L),
            Pair.of(createDummyJobCtx("job1", 3), 1020L)));
    jobs.put("job2", ImmutableSet.of(Pair.of(createDummyJobCtx("job2", 1), 1000L)));
    Pair<TableRebalanceContext, Long> jobTime = RebalanceChecker.getLatestJob(jobs);
    assertNotNull(jobTime);
    assertEquals(jobTime.getLeft().getJobId(), "job1_3");

    // The most recent job run is job1_4, but reached 3 maxAttempts.
    jobs.put("job1",
        ImmutableSet.of(Pair.of(createDummyJobCtx("job1", 1), 10L), Pair.of(createDummyJobCtx("job1", 2), 20L),
            Pair.of(createDummyJobCtx("job1", 3), 1020L), Pair.of(createDummyJobCtx("job1", 4), 2020L)));
    jobTime = RebalanceChecker.getLatestJob(jobs);
    assertNotNull(jobTime);
    assertEquals(jobTime.getLeft().getJobId(), "job2");

    // Add job3 that's started more recently.
    jobs.put("job3", ImmutableSet.of(Pair.of(createDummyJobCtx("job3", 1), 3000L)));
    jobTime = RebalanceChecker.getLatestJob(jobs);
    assertNotNull(jobTime);
    assertEquals(jobTime.getLeft().getJobId(), "job3");

    // Remove job2 and job3, and we'd have no job to retry then.
    jobs.remove("job2");
    jobs.remove("job3");
    jobTime = RebalanceChecker.getLatestJob(jobs);
    assertNull(jobTime);
  }