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

      Map<SubClusterId, SubClusterInfo> infoMap =
          this.gpgContext.getStateStoreFacade().getSubClusters(false, true);

      // Iterate over each sub cluster and check last heartbeat
      for (Map.Entry<SubClusterId, SubClusterInfo> entry : infoMap.entrySet()) {
        SubClusterInfo subClusterInfo = entry.getValue();

        Date lastHeartBeat = new Date(subClusterInfo.getLastHeartBeat());
        if (LOG.isDebugEnabled()) {
          LOG.debug("Checking subcluster {} in state {}, last heartbeat at {}",
              subClusterInfo.getSubClusterId(), subClusterInfo.getState(),
              lastHeartBeat);
        }

        if (subClusterInfo.getState().isUsable()) {
          long timeUntilDeregister = this.heartbeatExpirationMillis
              - (now.getTime() - lastHeartBeat.getTime());
          // Deregister sub-cluster as SC_LOST if last heartbeat too old
          if (timeUntilDeregister < 0) {
            LOG.warn(
                "Deregistering subcluster {} in state {} last heartbeat at {}",
                subClusterInfo.getSubClusterId(), subClusterInfo.getState(),
                new Date(subClusterInfo.getLastHeartBeat()));
            try {
              this.gpgContext.getStateStoreFacade().deregisterSubCluster(
                  subClusterInfo.getSubClusterId(), SubClusterState.SC_LOST);
            } catch (Exception e) {
              LOG.error("deregisterSubCluster failed on subcluster "
                  + subClusterInfo.getSubClusterId(), e);
            }
          } else if (LOG.isDebugEnabled()) {
            LOG.debug("Time until deregister for subcluster {}: {}",
                entry.getKey(),
                DurationFormatUtils.formatDurationISO(timeUntilDeregister));
          }
        }
      }
    } catch (Throwable e) {
      LOG.error("Subcluster cleaner fails: ", e);
    }
  }
@Test
  public void testSubClusterRegisterHeartBeatTime() throws YarnException {
    cleaner.run();
    Assert.assertEquals(3, facade.getSubClusters(true, true).size());
  }