@Override
  public void run() {
    try {
      // Step1. Get Current Time.
      Date now = new Date();
      LOG.info("SubClusterCleaner at {}.", now);

      Map<SubClusterId, SubClusterInfo> subClusters = federationFacade.getSubClusters(true);

      for (Map.Entry<SubClusterId, SubClusterInfo> subCluster : subClusters.entrySet()) {
        // Step2. Get information about subClusters.
        SubClusterId subClusterId = subCluster.getKey();
        SubClusterInfo subClusterInfo = subCluster.getValue();
        SubClusterState subClusterState = subClusterInfo.getState();
        long lastHeartBeatTime = subClusterInfo.getLastHeartBeat();

        // We Only Check SubClusters in NEW and RUNNING states
        if (subClusterState.isUsable()) {
          long heartBeatInterval = now.getTime() - lastHeartBeatTime;
          try {
            // HeartBeat Interval Exceeds Expiration Time
            if (heartBeatInterval > heartbeatExpirationMillis) {
              LOG.info("Deregister SubCluster {} in state {} last heartbeat at {}.",
                  subClusterId, subClusterState, new Date(lastHeartBeatTime));
              federationFacade.deregisterSubCluster(subClusterId, SubClusterState.SC_LOST);
            }
          } catch (YarnException e) {
            LOG.error("deregisterSubCluster failed on SubCluster {}.", subClusterId, e);
          }
        } else {
          LOG.debug("SubCluster {} in state {} last heartbeat at {}, " +
              "heartbeat interval < 30mins, no need for Deregister.",
              subClusterId, subClusterState, new Date(lastHeartBeatTime));
        }
      }
    } catch (Throwable e) {
      LOG.error("SubClusterCleaner Fails.", e);
    }
  }
@Test
  public void testSubClustersPartWithHeartBeat() throws YarnException, InterruptedException {

    // Step1. Manually set subCluster heartbeat expiration.
    for (int i = 0; i < NUM_SUBCLUSTERS; i++) {
      // Create subCluster id and info.
      expiredSubcluster("SC-" + i);
    }

    // Step2. Run the Cleaner to change the status of the expired SubCluster to SC_LOST.
    cleaner.run();

    // Step3. Let SC-0, SC-1 resume heartbeat.
    resumeSubClusterHeartbeat("SC-0");
    resumeSubClusterHeartbeat("SC-1");

    // Step4. At this point we should have 2 subClusters that are surviving clusters.
    int count = facade.getActiveSubClustersCount();
    Assert.assertEquals(2, count);

    // Step5. The result we expect is that SC-0 and SC-1 are in the RUNNING state,
    // and SC-2 and SC-3 are in the SC_LOST state.
    checkSubClusterState("SC-0", SubClusterState.SC_RUNNING);
    checkSubClusterState("SC-1", SubClusterState.SC_RUNNING);
    checkSubClusterState("SC-2", SubClusterState.SC_LOST);
    checkSubClusterState("SC-3", SubClusterState.SC_LOST);
  }