@Override
  public ClusterUserInfo getClusterUserInfo(HttpServletRequest hsr) {
    try {
      long startTime = Time.now();
      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("getClusterUserInfo", argsClasses, args);
      Map<SubClusterInfo, ClusterUserInfo> subClusterInfoMap =
          invokeConcurrent(subClustersActive, remoteMethod, ClusterUserInfo.class);
      FederationClusterUserInfo federationClusterUserInfo = new FederationClusterUserInfo();
      subClusterInfoMap.forEach((subClusterInfo, clusterUserInfo) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        clusterUserInfo.setSubClusterId(subClusterId.getId());
        federationClusterUserInfo.getList().add(clusterUserInfo);
      });
      long stopTime = Time.now();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), GET_CLUSTERUSERINFO,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededGetClusterUserInfoRetrieved(stopTime - startTime);
      return federationClusterUserInfo;
    } catch (NotFoundException e) {
      routerMetrics.incrGetClusterUserInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERUSERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("Get all active sub cluster(s) error.", e);
    } catch (YarnException | IOException e) {
      routerMetrics.incrGetClusterUserInfoFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERUSERINFO, UNKNOWN,
          TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("getClusterUserInfo error.", e);
    }
    routerMetrics.incrGetClusterUserInfoFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), GET_CLUSTERUSERINFO, UNKNOWN,
        TARGET_WEB_SERVICE, "getClusterUserInfo error.");
    throw new RuntimeException("getClusterUserInfo error.");
  }
@Test
  public void testGetClusterUserInfo() {
    String requestUserName = "test-user";
    HttpServletRequest hsr = mock(HttpServletRequest.class);
    when(hsr.getRemoteUser()).thenReturn(requestUserName);
    ClusterUserInfo clusterUserInfo = interceptor.getClusterUserInfo(hsr);

    Assert.assertNotNull(clusterUserInfo);
    Assert.assertTrue(clusterUserInfo instanceof FederationClusterUserInfo);

    FederationClusterUserInfo federationClusterUserInfo =
        (FederationClusterUserInfo) clusterUserInfo;

    List<ClusterUserInfo> fedClusterUserInfoList = federationClusterUserInfo.getList();
    Assert.assertNotNull(fedClusterUserInfoList);
    Assert.assertEquals(4, fedClusterUserInfoList.size());

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

    for (ClusterUserInfo fedClusterUserInfo : fedClusterUserInfoList) {
      // Check subClusterId
      String subClusterId = fedClusterUserInfo.getSubClusterId();
      Assert.assertNotNull(subClusterId);
      Assert.assertTrue(subClusterIds.contains(subClusterId));

      // Check requestedUser
      String requestedUser = fedClusterUserInfo.getRequestedUser();
      Assert.assertNotNull(requestedUser);
      Assert.assertEquals(requestUserName, requestedUser);

      // Check rmLoginUser
      String rmLoginUser = fedClusterUserInfo.getRmLoginUser();
      Assert.assertNotNull(rmLoginUser);
      Assert.assertEquals(mockRM.getRMLoginUser(), rmLoginUser);
    }
  }