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

    FederationStateStoreFacade facade = getGPGContext().getStateStoreFacade();
    try {
      // Get the candidate list from StateStore before calling router
      Set<ApplicationId> allStateStoreApps = new HashSet<>();
      List<ApplicationHomeSubCluster> response =
          facade.getApplicationsHomeSubCluster();
      for (ApplicationHomeSubCluster app : response) {
        allStateStoreApps.add(app.getApplicationId());
      }
      LOG.info("{} app entries in FederationStateStore", allStateStoreApps.size());

      // Get the candidate list from Registry before calling router
      List<String> allRegistryApps = getRegistryClient().getAllApplications();
      LOG.info("{} app entries in FederationRegistry", allStateStoreApps.size());

      // Get the list of known apps from Router
      Set<ApplicationId> routerApps = getRouterKnownApplications();
      LOG.info("{} known applications from Router", routerApps.size());

      // Clean up StateStore entries
      Set<ApplicationId> toDelete =
          Sets.difference(allStateStoreApps, routerApps);

      LOG.info("Deleting {} applications from statestore", toDelete.size());
      LOG.debug("Apps to delete: {}.",
          toDelete.stream().map(Object::toString).collect(Collectors.joining(",")));

      for (ApplicationId appId : toDelete) {
        try {
          LOG.debug("Deleting {} from statestore ", appId);
          facade.deleteApplicationHomeSubCluster(appId);
        } catch (Exception e) {
          LOG.error("deleteApplicationHomeSubCluster failed at application {}.", appId, e);
        }
      }

      // Clean up Registry entries
      for (String app : allRegistryApps) {
        ApplicationId appId = ApplicationId.fromString(app);
        if (!routerApps.contains(appId)) {
          LOG.debug("removing finished application entry for {}", app);
          getRegistryClient().removeAppFromRegistry(appId, true);
        }
      }
    } catch (Throwable e) {
      LOG.error("Application cleaner started at time {} fails. ", now, e);
    }
  }
@Test
  public void testConcurrentNewApp() throws YarnException {
    appIdToAddConcurrently = ApplicationId.newInstance(1, 1);

    appCleaner.run();

    // The concurrently added app should be still there
    GetApplicationsHomeSubClusterRequest appHomeSubClusterRequest =
         GetApplicationsHomeSubClusterRequest.newInstance();
    GetApplicationsHomeSubClusterResponse applicationsHomeSubCluster =
        stateStore.getApplicationsHomeSubCluster(appHomeSubClusterRequest);
    Assert.assertNotNull(applicationsHomeSubCluster);
    List<ApplicationHomeSubCluster> appsHomeSubClusters =
        applicationsHomeSubCluster.getAppsHomeSubClusters();
    Assert.assertNotNull(appsHomeSubClusters);
    Assert.assertEquals(1, appsHomeSubClusters.size());

    // The concurrently added app should be still there
    Assert.assertEquals(1, registryClient.getAllApplications().size());
  }