@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 testSubClustersWithOutHeartBeat()
      throws InterruptedException, TimeoutException, YarnException {

    // We set up such a unit test, We set the status of all subClusters to RUNNING,
    // and Manually set subCluster heartbeat expiration.
    // At this time, the size of the Active SubCluster is 0.
    Map<SubClusterId, SubClusterInfo> subClustersMap = facade.getSubClusters(false);

    // Step1. Manually set subCluster heartbeat expiration.
    // subCluster has no heartbeat, and all subClusters will expire.
    subClustersMap.keySet().forEach(subClusterId ->
        stateStore.setExpiredHeartbeat(subClusterId, EXPIRATION_TIME));

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

    // Step3. All clusters have expired,
    // so the current Federation has no active subClusters.
    int count = facade.getActiveSubClustersCount();
    Assert.assertEquals(0, count);

    // Step4. Check Active SubCluster Status.
    // We want all subClusters to be SC_LOST.
    subClustersMap.values().forEach(subClusterInfo -> {
      SubClusterState subClusterState = subClusterInfo.getState();
      Assert.assertEquals(SubClusterState.SC_LOST, subClusterState);
    });
  }