@Override
  public SchedulerTypeInfo getSchedulerInfo() {
    try {
      long startTime = Time.now();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      Class[] argsClasses = new Class[]{};
      Object[] args = new Object[]{};
      ClientMethod remoteMethod = new ClientMethod("getSchedulerInfo", argsClasses, args);
      Map<SubClusterInfo, SchedulerTypeInfo> subClusterInfoMap =
          invokeConcurrent(subClustersActive, remoteMethod, SchedulerTypeInfo.class);
      FederationSchedulerTypeInfo federationSchedulerTypeInfo = new FederationSchedulerTypeInfo();
      subClusterInfoMap.forEach((subClusterInfo, schedulerTypeInfo) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        schedulerTypeInfo.setSubClusterId(subClusterId.getId());
        federationSchedulerTypeInfo.getList().add(schedulerTypeInfo);
      });
      long stopTime = Time.now();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), GET_SCHEDULERINFO,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededGetSchedulerInfoRetrieved(stopTime - startTime);
      return federationSchedulerTypeInfo;
    } catch (NotFoundException e) {
      routerMetrics.incrGetSchedulerInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("Get all active sub cluster(s) error.", e);
    } catch (YarnException | IOException e) {
      routerMetrics.incrGetSchedulerInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("getSchedulerInfo error.", e);
    }
    routerMetrics.incrGetSchedulerInfoFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULERINFO, UNKNOWN,
        TARGET_WEB_SERVICE, "getSchedulerInfo error.");
    throw new RuntimeException("getSchedulerInfo error.");
  }
@Test
  public void testGetSchedulerInfo() {
    // In this test case, we will get the return results of 4 sub-clusters.
    SchedulerTypeInfo typeInfo = interceptor.getSchedulerInfo();
    Assert.assertNotNull(typeInfo);
    Assert.assertTrue(typeInfo instanceof FederationSchedulerTypeInfo);

    FederationSchedulerTypeInfo federationSchedulerTypeInfo =
        (FederationSchedulerTypeInfo) typeInfo;
    Assert.assertNotNull(federationSchedulerTypeInfo);
    List<SchedulerTypeInfo> schedulerTypeInfos = federationSchedulerTypeInfo.getList();
    Assert.assertNotNull(schedulerTypeInfos);
    Assert.assertEquals(4, schedulerTypeInfos.size());
    List<String> subClusterIds = subClusters.stream().map(SubClusterId::getId).
        collect(Collectors.toList());

    for (SchedulerTypeInfo schedulerTypeInfo : schedulerTypeInfos) {
      Assert.assertNotNull(schedulerTypeInfo);

      // 1. Whether the returned subClusterId is in the subCluster list
      String subClusterId = schedulerTypeInfo.getSubClusterId();
      Assert.assertTrue(subClusterIds.contains(subClusterId));

      // 2. We test CapacityScheduler, the returned type should be CapacityScheduler.
      SchedulerInfo schedulerInfo = schedulerTypeInfo.getSchedulerInfo();
      Assert.assertNotNull(schedulerInfo);
      Assert.assertTrue(schedulerInfo instanceof CapacitySchedulerInfo);
      CapacitySchedulerInfo capacitySchedulerInfo = (CapacitySchedulerInfo) schedulerInfo;
      Assert.assertNotNull(capacitySchedulerInfo);

      // 3. The parent queue name should be root
      String queueName = capacitySchedulerInfo.getQueueName();
      Assert.assertEquals("root", queueName);

      // 4. schedulerType should be CapacityScheduler
      String schedulerType = capacitySchedulerInfo.getSchedulerType();
      Assert.assertEquals("Capacity Scheduler", schedulerType);

      // 5. queue path should be root
      String queuePath = capacitySchedulerInfo.getQueuePath();
      Assert.assertEquals("root", queuePath);

      // 6. mockRM has 2 test queues, [root.a, root.b]
      List<String> queues = Lists.newArrayList("root.a", "root.b");
      CapacitySchedulerQueueInfoList csSchedulerQueueInfoList = capacitySchedulerInfo.getQueues();
      Assert.assertNotNull(csSchedulerQueueInfoList);
      List<CapacitySchedulerQueueInfo> csQueueInfoList =
          csSchedulerQueueInfoList.getQueueInfoList();
      Assert.assertEquals(2, csQueueInfoList.size());
      for (CapacitySchedulerQueueInfo csQueueInfo : csQueueInfoList) {
        Assert.assertNotNull(csQueueInfo);
        Assert.assertTrue(queues.contains(csQueueInfo.getQueuePath()));
      }
    }
  }