@Override
  public ClusterMetricsInfo getClusterMetricsInfo() {
    ClusterMetricsInfo metrics = new ClusterMetricsInfo();

    Collection<SubClusterInfo> subClusterInfos = federationFacade.getActiveSubClusters();

    Stream<ClusterMetricsInfo> clusterMetricsInfoStream = subClusterInfos.parallelStream()
        .map(subClusterInfo -> {
          DefaultRequestInterceptorREST interceptor =
              getOrCreateInterceptorForSubCluster(subClusterInfo);
          try {
            return interceptor.getClusterMetricsInfo();
          } catch (Exception e) {
            LOG.error("Subcluster {} failed to return Cluster Metrics.",
                subClusterInfo.getSubClusterId());
            return null;
          }
        });

    clusterMetricsInfoStream.forEach(clusterMetricsInfo -> {
      try {
        if (clusterMetricsInfo != null) {
          RouterWebServiceUtil.mergeMetrics(metrics, clusterMetricsInfo);
        }
      } catch (Throwable e) {
        LOG.warn("Failed to get nodes report.", e);
      }
    });

    return metrics;
  }
@Test
  public void testGetClusterMetrics() {

    ClusterMetricsInfo responseGet = interceptor.getClusterMetricsInfo();

    Assert.assertNotNull(responseGet);
    int expectedAppSubmitted = 0;
    for (int i = 0; i < NUM_SUBCLUSTER; i++) {
      expectedAppSubmitted += i;
    }
    Assert.assertEquals(expectedAppSubmitted, responseGet.getAppsSubmitted());
    // The merge operations is tested in TestRouterWebServiceUtil
  }