@GET
  @Path(RMWSConsts.CHECK_USER_ACCESS_TO_QUEUE)
  @Produces({ MediaType.APPLICATION_JSON + "; " + JettyUtils.UTF_8,
                MediaType.APPLICATION_XML + "; " + JettyUtils.UTF_8 })
  public RMQueueAclInfo checkUserAccessToQueue(
      @PathParam(RMWSConsts.QUEUE) String queue,
      @QueryParam(RMWSConsts.USER) String username,
      @QueryParam(RMWSConsts.QUEUE_ACL_TYPE)
        @DefaultValue("SUBMIT_APPLICATIONS") String queueAclType,
      @Context HttpServletRequest hsr) throws AuthorizationException {
    initForReadableEndpoints();

    // For the user who invokes this REST call, he/she should have admin access
    // to the queue. Otherwise, we will reject the call.
    UserGroupInformation callerUGI = getCallerUserGroupInformation(hsr, true);
    if (callerUGI != null && !this.rm.getResourceScheduler().checkAccess(
        callerUGI, QueueACL.ADMINISTER_QUEUE, queue)) {
      throw new ForbiddenException(
          "User=" + callerUGI.getUserName() + " doesn't haven access to queue="
              + queue + " so it cannot check ACLs for other users.");
    }

    // Create UGI for the to-be-checked user.
    UserGroupInformation user = UserGroupInformation.createRemoteUser(username);
    if (user == null) {
      throw new ForbiddenException(
          "Failed to retrieve UserGroupInformation for user=" + username);
    }

    // Check if the specified queue acl is valid.
    QueueACL queueACL;
    try {
      queueACL = QueueACL.valueOf(queueAclType);
    } catch (IllegalArgumentException e) {
      throw new BadRequestException("Specified queueAclType=" + queueAclType
          + " is not a valid type, valid queue acl types={"
          + "SUBMIT_APPLICATIONS/ADMINISTER_QUEUE}");
    }

    if (!this.rm.getResourceScheduler().checkAccess(user, queueACL, queue)) {
      return new RMQueueAclInfo(false, user.getUserName(),
          "User=" + username + " doesn't have access to queue=" + queue
              + " with acl-type=" + queueAclType);
    }

    return new RMQueueAclInfo(true, user.getUserName(), "");
  }
@Test
  public void testCheckUserAccessToQueue() throws Exception {

    ResourceManager mockRM = mock(ResourceManager.class);
    Configuration conf = new YarnConfiguration();

    // Inject a mock scheduler implementation.
    // Only admin user has ADMINISTER_QUEUE access.
    // For SUBMIT_APPLICATION ACL, both of admin/yarn user have acess
    ResourceScheduler mockScheduler = new FifoScheduler() {
      @Override
      public synchronized boolean checkAccess(UserGroupInformation callerUGI,
          QueueACL acl, String queueName) {
        if (acl == QueueACL.ADMINISTER_QUEUE) {
          if (callerUGI.getUserName().equals("admin")) {
            return true;
          }
        } else {
          if (ImmutableSet.of("admin", "yarn").contains(callerUGI.getUserName())) {
            return true;
          }
        }
        return false;
      }
    };

    when(mockRM.getResourceScheduler()).thenReturn(mockScheduler);

    RMWebServices webSvc =
        new RMWebServices(mockRM, conf, mock(HttpServletResponse.class));

    boolean caughtException = false;

    // Case 1: Only queue admin user can access other user's information
    HttpServletRequest mockHsr = mockHttpServletRequestByUserName("non-admin");
    try {
      webSvc.checkUserAccessToQueue("queue", "jack",
          QueueACL.SUBMIT_APPLICATIONS.name(), mockHsr);
    } catch (ForbiddenException e) {
      caughtException = true;
    }
    Assert.assertTrue(caughtException);

    // Case 2: request an unknown ACL causes BAD_REQUEST
    mockHsr = mockHttpServletRequestByUserName("admin");
    caughtException = false;
    try {
      webSvc.checkUserAccessToQueue("queue", "jack", "XYZ_ACL", mockHsr);
    } catch (BadRequestException e) {
      caughtException = true;
    }
    Assert.assertTrue(caughtException);

    // Case 3: get FORBIDDEN for rejected ACL
    mockHsr = mockHttpServletRequestByUserName("admin");
    Assert.assertFalse(webSvc.checkUserAccessToQueue("queue", "jack",
        QueueACL.SUBMIT_APPLICATIONS.name(), mockHsr).isAllowed());
    Assert.assertFalse(webSvc.checkUserAccessToQueue("queue", "jack",
        QueueACL.ADMINISTER_QUEUE.name(), mockHsr).isAllowed());

    // Case 4: get OK for listed ACLs
    mockHsr = mockHttpServletRequestByUserName("admin");
    Assert.assertTrue(webSvc.checkUserAccessToQueue("queue", "admin",
        QueueACL.SUBMIT_APPLICATIONS.name(), mockHsr).isAllowed());
    Assert.assertTrue(webSvc.checkUserAccessToQueue("queue", "admin",
        QueueACL.ADMINISTER_QUEUE.name(), mockHsr).isAllowed());

    // Case 5: get OK only for SUBMIT_APP acl for "yarn" user
    mockHsr = mockHttpServletRequestByUserName("admin");
    Assert.assertTrue(webSvc.checkUserAccessToQueue("queue", "yarn",
        QueueACL.SUBMIT_APPLICATIONS.name(), mockHsr).isAllowed());
    Assert.assertFalse(webSvc.checkUserAccessToQueue("queue", "yarn",
        QueueACL.ADMINISTER_QUEUE.name(), mockHsr).isAllowed());
  }