@Override
  public Response getSchedulerConfiguration(HttpServletRequest hsr)
      throws AuthorizationException {
    try {
      long startTime = clock.getTime();
      FederationConfInfo federationConfInfo = new FederationConfInfo();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      final HttpServletRequest hsrCopy = clone(hsr);
      Class[] argsClasses = new Class[]{HttpServletRequest.class};
      Object[] args = new Object[]{hsrCopy};
      ClientMethod remoteMethod = new ClientMethod("getSchedulerConfiguration", argsClasses, args);
      Map<SubClusterInfo, Response> responseMap =
          invokeConcurrent(subClustersActive, remoteMethod, Response.class);
      responseMap.forEach((subClusterInfo, response) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        if (response == null) {
          String errorMsg = subClusterId + " Can't getSchedulerConfiguration.";
          federationConfInfo.getErrorMsgs().add(errorMsg);
        } else if (response.getStatus() == Status.BAD_REQUEST.getStatusCode()) {
          String errorMsg = String.valueOf(response.getEntity());
          federationConfInfo.getErrorMsgs().add(errorMsg);
        } else if (response.getStatus() == Status.OK.getStatusCode()) {
          ConfInfo fedConfInfo = (ConfInfo) response.getEntity();
          fedConfInfo.setSubClusterId(subClusterId.getId());
          federationConfInfo.getList().add(fedConfInfo);
        }
      });
      long endTime = clock.getTime();
      routerMetrics.succeededGetSchedulerConfigurationRetrieved(endTime - startTime);
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), GET_SCHEDULER_CONFIGURATION,
          TARGET_WEB_SERVICE);
      return Response.status(Status.OK).entity(federationConfInfo).build();
    } catch (NotFoundException e) {
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      routerMetrics.incrGetSchedulerConfigurationFailedRetrieved();
      RouterServerUtil.logAndThrowRunTimeException("get all active sub cluster(s) error.", e);
    } catch (Exception e) {
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULER_CONFIGURATION,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      routerMetrics.incrGetSchedulerConfigurationFailedRetrieved();
      return Response.status(Status.BAD_REQUEST).entity("getSchedulerConfiguration error.").build();
    }

    routerMetrics.incrGetSchedulerConfigurationFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_SCHEDULER_CONFIGURATION,
        UNKNOWN, TARGET_WEB_SERVICE, "getSchedulerConfiguration Failed.");
    throw new RuntimeException("getSchedulerConfiguration Failed.");
  }
@Test
  public void testGetSchedulerConfiguration() throws Exception {
    Response response = interceptor.getSchedulerConfiguration(null);
    Assert.assertNotNull(response);
    Assert.assertEquals(OK, response.getStatus());

    Object entity = response.getEntity();
    Assert.assertNotNull(entity);
    Assert.assertTrue(entity instanceof FederationConfInfo);

    FederationConfInfo federationConfInfo = FederationConfInfo.class.cast(entity);
    List<ConfInfo> confInfos = federationConfInfo.getList();
    Assert.assertNotNull(confInfos);
    Assert.assertEquals(4, confInfos.size());

    List<String> errors = federationConfInfo.getErrorMsgs();
    Assert.assertEquals(0, errors.size());

    Set<String> subClusterSet = subClusters.stream()
        .map(subClusterId -> subClusterId.getId()).collect(Collectors.toSet());

    for (ConfInfo confInfo : confInfos) {
      List<ConfInfo.ConfItem> confItems = confInfo.getItems();
      Assert.assertNotNull(confItems);
      Assert.assertTrue(confItems.size() > 0);
      Assert.assertTrue(subClusterSet.contains(confInfo.getSubClusterId()));
    }
  }