@Override
  public RMQueueAclInfo checkUserAccessToQueue(String queue, String username,
      String queueAclType, HttpServletRequest hsr) throws AuthorizationException {

    // Parameter Verification
    if (queue == null || queue.isEmpty()) {
      routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          UNKNOWN, TARGET_WEB_SERVICE, "Parameter error, the queue is empty or null.");
      throw new IllegalArgumentException("Parameter error, the queue is empty or null.");
    }

    if (username == null || username.isEmpty()) {
      routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          UNKNOWN, TARGET_WEB_SERVICE, "Parameter error, the username is empty or null.");
      throw new IllegalArgumentException("Parameter error, the username is empty or null.");
    }

    if (queueAclType == null || queueAclType.isEmpty()) {
      routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          UNKNOWN, TARGET_WEB_SERVICE, "Parameter error, the queueAclType is empty or null.");
      throw new IllegalArgumentException("Parameter error, the queueAclType is empty or null.");
    }

    // Traverse SubCluster and call checkUserAccessToQueue Api
    try {
      long startTime = Time.now();
      Collection<SubClusterInfo> subClustersActive = federationFacade.getActiveSubClusters();
      final HttpServletRequest hsrCopy = clone(hsr);
      Class[] argsClasses = new Class[]{String.class, String.class, String.class,
          HttpServletRequest.class};
      Object[] args = new Object[]{queue, username, queueAclType, hsrCopy};
      ClientMethod remoteMethod = new ClientMethod("checkUserAccessToQueue", argsClasses, args);
      Map<SubClusterInfo, RMQueueAclInfo> rmQueueAclInfoMap =
          invokeConcurrent(subClustersActive, remoteMethod, RMQueueAclInfo.class);
      FederationRMQueueAclInfo aclInfo = new FederationRMQueueAclInfo();
      rmQueueAclInfoMap.forEach((subClusterInfo, rMQueueAclInfo) -> {
        SubClusterId subClusterId = subClusterInfo.getSubClusterId();
        rMQueueAclInfo.setSubClusterId(subClusterId.getId());
        aclInfo.getList().add(rMQueueAclInfo);
      });
      long stopTime = Time.now();
      RouterAuditLogger.logSuccess(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          TARGET_WEB_SERVICE);
      routerMetrics.succeededCheckUserAccessToQueueRetrieved(stopTime - startTime);
      return aclInfo;
    } catch (NotFoundException e) {
      routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("Get all active sub cluster(s) error.", e);
    } catch (YarnException | IOException e) {
      routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
      RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
          UNKNOWN, TARGET_WEB_SERVICE, e.getLocalizedMessage());
      RouterServerUtil.logAndThrowRunTimeException("checkUserAccessToQueue error.", e);
    }

    routerMetrics.incrCheckUserAccessToQueueFailedRetrieved();
    RouterAuditLogger.logFailure(getUser().getShortUserName(), CHECK_USER_ACCESS_TO_QUEUE,
        UNKNOWN, TARGET_WEB_SERVICE, "checkUserAccessToQueue error.");
    throw new RuntimeException("checkUserAccessToQueue error.");
  }
@Test
  public void testCheckUserAccessToQueue() throws Exception {

    interceptor.setAllowPartialResult(false);

    // Case 1: Only queue admin user can access other user's information
    HttpServletRequest mockHsr = mockHttpServletRequestByUserName("non-admin");
    String errorMsg1 = "User=non-admin doesn't haven access to queue=queue " +
        "so it cannot check ACLs for other users.";
    LambdaTestUtils.intercept(YarnRuntimeException.class, errorMsg1,
        () -> interceptor.checkUserAccessToQueue("queue", "jack",
        QueueACL.SUBMIT_APPLICATIONS.name(), mockHsr));

    // Case 2: request an unknown ACL causes BAD_REQUEST
    HttpServletRequest mockHsr1 = mockHttpServletRequestByUserName("admin");
    String errorMsg2 = "Specified queueAclType=XYZ_ACL is not a valid type, " +
        "valid queue acl types={SUBMIT_APPLICATIONS/ADMINISTER_QUEUE}";
    LambdaTestUtils.intercept(YarnRuntimeException.class, errorMsg2,
        () -> interceptor.checkUserAccessToQueue("queue", "jack", "XYZ_ACL", mockHsr1));

    // We design a test, admin user has ADMINISTER_QUEUE, SUBMIT_APPLICATIONS permissions,
    // yarn user has SUBMIT_APPLICATIONS permissions, other users have no permissions

    // Case 3: get FORBIDDEN for rejected ACL
    checkUserAccessToQueueFailed("queue", "jack", QueueACL.SUBMIT_APPLICATIONS, "admin");
    checkUserAccessToQueueFailed("queue", "jack", QueueACL.ADMINISTER_QUEUE, "admin");

    // Case 4: get OK for listed ACLs
    checkUserAccessToQueueSuccess("queue", "admin", QueueACL.ADMINISTER_QUEUE, "admin");
    checkUserAccessToQueueSuccess("queue", "admin", QueueACL.SUBMIT_APPLICATIONS, "admin");

    // Case 5: get OK only for SUBMIT_APP acl for "yarn" user
    checkUserAccessToQueueFailed("queue", "yarn", QueueACL.ADMINISTER_QUEUE, "admin");
    checkUserAccessToQueueSuccess("queue", "yarn", QueueACL.SUBMIT_APPLICATIONS, "admin");

    interceptor.setAllowPartialResult(true);
  }