@Override
  public ApplicationStatisticsInfo getAppStatistics(HttpServletRequest hsr,
      Set<String> stateQueries, Set<String> typeQueries) {
    try {
      long startTime = clock.getTime();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      final HttpServletRequest hsrCopy = clone(hsr);
      Class[] argsClasses = new Class[]{HttpServletRequest.class, Set.class, Set.class};
      Object[] args = new Object[]{hsrCopy, stateQueries, typeQueries};
      ClientMethod remoteMethod = new ClientMethod("getAppStatistics", argsClasses, args);
      Map<SubClusterInfo, ApplicationStatisticsInfo> appStatisticsMap = invokeConcurrent(
          subClustersActive, remoteMethod, ApplicationStatisticsInfo.class);
      ApplicationStatisticsInfo applicationStatisticsInfo  =
          RouterWebServiceUtil.mergeApplicationStatisticsInfo(appStatisticsMap.values());
      if (applicationStatisticsInfo != null) {
        long stopTime = clock.getTime();
        routerMetrics.succeededGetAppStatisticsRetrieved(stopTime - startTime);
        RouterAuditLogger.logSuccess(getUser().getShortUserName(), GET_APPSTATISTICS,
            TARGET_WEB_SERVICE);
        return applicationStatisticsInfo;
      }
    } catch (NotFoundException e) {
      routerMetrics.incrGetAppStatisticsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_APPSTATISTICS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("get all active sub cluster(s) error.", e);
    } catch (IOException e) {
      routerMetrics.incrGetAppStatisticsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_APPSTATISTICS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "getAppStatistics error by stateQueries = %s, typeQueries = %s with io error.",
          StringUtils.join(stateQueries, ","), StringUtils.join(typeQueries, ","));
    } catch (YarnException e) {
      routerMetrics.incrGetAppStatisticsFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_APPSTATISTICS, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException(e,
          "getAppStatistics by stateQueries = %s, typeQueries = %s with yarn error.",
          StringUtils.join(stateQueries, ","), StringUtils.join(typeQueries, ","));
    }
    routerMetrics.incrGetAppStatisticsFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_APPSTATISTICS, UNKNOWN,
        TARGET_WEB_SERVICE, "getAppStatistics Failed.");
    throw RouterServerUtil.logAndReturnRunTimeException(
        "getAppStatistics by stateQueries = %s, typeQueries = %s Failed.",
        StringUtils.join(stateQueries, ","), StringUtils.join(typeQueries, ","));
  }
@Test
  public void testGetAppStatistics() throws IOException, InterruptedException, YarnException {

    // Submit application to multiSubCluster
    ApplicationId appId = ApplicationId.newInstance(Time.now(), 1);
    ApplicationSubmissionContextInfo context = new ApplicationSubmissionContextInfo();
    context.setApplicationId(appId.toString());
    context.setApplicationType("MapReduce");
    context.setQueue("queue");

    Assert.assertNotNull(interceptor.submitApplication(context, null));

    GetApplicationHomeSubClusterRequest request =
        GetApplicationHomeSubClusterRequest.newInstance(appId);
    GetApplicationHomeSubClusterResponse response =
        stateStore.getApplicationHomeSubCluster(request);

    Assert.assertNotNull(response);
    ApplicationHomeSubCluster homeSubCluster = response.getApplicationHomeSubCluster();

    DefaultRequestInterceptorREST interceptorREST =
        interceptor.getInterceptorForSubCluster(homeSubCluster.getHomeSubCluster());

    MockDefaultRequestInterceptorREST mockInterceptorREST =
        (MockDefaultRequestInterceptorREST) interceptorREST;
    mockInterceptorREST.updateApplicationState(YarnApplicationState.RUNNING,
        appId.toString());

    Set<String> stateQueries = new HashSet<>();
    stateQueries.add(YarnApplicationState.RUNNING.name());

    Set<String> typeQueries = new HashSet<>();
    typeQueries.add("MapReduce");

    ApplicationStatisticsInfo response2 =
        interceptor.getAppStatistics(null, stateQueries, typeQueries);

    Assert.assertNotNull(response2);
    Assert.assertFalse(response2.getStatItems().isEmpty());

    StatisticsItemInfo result = response2.getStatItems().get(0);
    Assert.assertEquals(1, result.getCount());
    Assert.assertEquals(YarnApplicationState.RUNNING, result.getState());
    Assert.assertEquals("MapReduce", result.getType());
  }