@Override
  public ClusterInfo getClusterInfo() {
    try {
      long startTime = Time.now();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      Class[] argsClasses = new Class[]{};
      Object[] args = new Object[]{};
      ClientMethod remoteMethod = new ClientMethod("getClusterInfo", argsClasses, args);
      Map<SubClusterInfo, ClusterInfo> subClusterInfoMap =
          invokeConcurrent(subClustersActive, remoteMethod, ClusterInfo.class);
      FederationClusterInfo federationClusterInfo = new FederationClusterInfo();
      subClusterInfoMap.forEach((subClusterInfo, clusterInfo) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        clusterInfo.setSubClusterId(subClusterId.getId());
        federationClusterInfo.getList().add(clusterInfo);
      });
      long stopTime = Time.now();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), GET_CLUSTERINFO,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededGetClusterInfoRetrieved(stopTime - startTime);
      return federationClusterInfo;
    } catch (NotFoundException e) {
      routerMetrics.incrGetClusterInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("Get all active sub cluster(s) error.", e);
    } catch (YarnException | IOException e) {
      routerMetrics.incrGetClusterInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("getClusterInfo error.", e);
    }
    routerMetrics.incrGetClusterInfoFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERINFO, UNKNOWN,
        TARGET_WEB_SERVICE, "getClusterInfo error.");
    throw new RuntimeException("getClusterInfo error.");
  }
@Test
  public void testGetClusterInfo() {
    ClusterInfo clusterInfos = interceptor.getClusterInfo();
    Assert.assertNotNull(clusterInfos);
    Assert.assertTrue(clusterInfos instanceof FederationClusterInfo);

    FederationClusterInfo federationClusterInfos =
        (FederationClusterInfo) (clusterInfos);

    List<ClusterInfo> fedClusterInfosList = federationClusterInfos.getList();
    Assert.assertNotNull(fedClusterInfosList);
    Assert.assertEquals(4, fedClusterInfosList.size());

    List<String> subClusterIds = subClusters.stream().map(
        subClusterId -> subClusterId.getId()).collect(Collectors.toList());

    MockRM mockRM = interceptor.getMockRM();
    String yarnVersion = YarnVersionInfo.getVersion();

    for (ClusterInfo clusterInfo : fedClusterInfosList) {
      String subClusterId = clusterInfo.getSubClusterId();
      // Check subClusterId
      Assert.assertTrue(subClusterIds.contains(subClusterId));

      // Check state
      String clusterState = mockRM.getServiceState().toString();
      Assert.assertEquals(clusterState, clusterInfo.getState());

      // Check rmStateStoreName
      String rmStateStoreName =
          mockRM.getRMContext().getStateStore().getClass().getName();
      Assert.assertEquals(rmStateStoreName, clusterInfo.getRMStateStore());

      // Check RM Version
      Assert.assertEquals(yarnVersion, clusterInfo.getRMVersion());

      // Check haZooKeeperConnectionState
      String rmHAZookeeperConnectionState = mockRM.getRMContext().getHAZookeeperConnectionState();
      Assert.assertEquals(rmHAZookeeperConnectionState,
          clusterInfo.getHAZookeeperConnectionState());
    }
  }