YARN-899. Added back queue level administrator-acls so that there is no regression w.r.t 1.x. Contributed by Xuan Gong.

svn merge --ignore-ancestry -c 1527282 ../../trunk/


git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-2@1527283 13f79535-47bb-0310-9956-ffa450edef68

hadoop-tools/hadoop-sls/src/main/java/org/apache/hadoop/yarn/sls/scheduler/ResourceSchedulerWrapper.java

@@ -29,6 +29,7 @@ import com.codahale.metrics.MetricRegistry;
 import com.codahale.metrics.SlidingWindowReservoir;
 import com.codahale.metrics.Timer;
 
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.conf.Configurable;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.util.ReflectionUtils;
@@ -41,6 +42,7 @@ import org.apache.hadoop.yarn.api.records.ContainerStatus;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.ResourceRequest;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
@@ -852,4 +854,11 @@ public class ResourceSchedulerWrapper implements ResourceScheduler,
   public QueueMetrics getRootQueueMetrics() {
     return scheduler.getRootQueueMetrics();
   }
+
+  @Override
+  public synchronized boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName) {
+    return scheduler.checkAccess(callerUGI, acl, queueName);
+  }
 }
+

hadoop-yarn-project/CHANGES.txt

@@ -64,6 +64,9 @@ Release 2.1.2 - UNRELEASED
     of testing given ApplicationHistoryServer is not yet ready. (Arpit Gupta via
     vinodkv)
 
+    YARN-899. Added back queue level administrator-acls so that there is no
+    regression w.r.t 1.x. (Xuan Gong via vinodkv)
+
   OPTIMIZATIONS
 
   BUG FIXES

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java

@@ -71,6 +71,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
@@ -90,6 +91,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.authorize.RMPolicyProvider;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
@@ -119,15 +121,18 @@ public class ClientRMService extends AbstractService implements
   InetSocketAddress clientBindAddress;
 
   private final ApplicationACLsManager applicationsACLsManager;
+  private final QueueACLsManager queueACLsManager;
 
   public ClientRMService(RMContext rmContext, YarnScheduler scheduler,
       RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
+      QueueACLsManager queueACLsManager,
       RMDelegationTokenSecretManager rmDTSecretManager) {
     super(ClientRMService.class.getName());
     this.scheduler = scheduler;
     this.rmContext = rmContext;
     this.rmAppManager = rmAppManager;
     this.applicationsACLsManager = applicationACLsManager;
+    this.queueACLsManager = queueACLsManager;
     this.rmDTSecretManager = rmDTSecretManager;
   }
 
@@ -185,13 +190,16 @@ public class ClientRMService extends AbstractService implements
    * @param callerUGI
    * @param owner
    * @param operationPerformed
-   * @param applicationId
+   * @param application
    * @return
    */
   private boolean checkAccess(UserGroupInformation callerUGI, String owner,
-      ApplicationAccessType operationPerformed, ApplicationId applicationId) {
+      ApplicationAccessType operationPerformed,
+      RMApp application) {
     return applicationsACLsManager.checkAccess(callerUGI, operationPerformed,
-        owner, applicationId);
+        owner, application.getApplicationId())
+        || queueACLsManager.checkAccess(callerUGI, QueueACL.ADMINISTER_QUEUE,
+            application.getQueue());
   }
 
   ApplicationId getNewApplicationId() {
@@ -241,7 +249,7 @@ public class ClientRMService extends AbstractService implements
     }
 
     boolean allowAccess = checkAccess(callerUGI, application.getUser(),
-        ApplicationAccessType.VIEW_APP, applicationId);
+        ApplicationAccessType.VIEW_APP, application);
     ApplicationReport report =
         application.createAndGetApplicationReport(callerUGI.getUserName(),
             allowAccess);
@@ -357,7 +365,7 @@ public class ClientRMService extends AbstractService implements
     }
 
     if (!checkAccess(callerUGI, application.getUser(),
-        ApplicationAccessType.MODIFY_APP, applicationId)) {
+        ApplicationAccessType.MODIFY_APP, application)) {
       RMAuditLogger.logFailure(callerUGI.getShortUserName(),
           AuditConstants.KILL_APP_REQUEST,
           "User doesn't have permissions to "
@@ -420,7 +428,7 @@ public class ClientRMService extends AbstractService implements
         }
       }
       boolean allowAccess = checkAccess(callerUGI, application.getUser(),
-          ApplicationAccessType.VIEW_APP, application.getApplicationId());
+          ApplicationAccessType.VIEW_APP, application);
       reports.add(application.createAndGetApplicationReport(
           callerUGI.getUserName(), allowAccess));
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ResourceManager.java

@@ -78,6 +78,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.security.AMRMTokenSecretMan
 import org.apache.hadoop.yarn.server.resourcemanager.security.ClientToAMTokenSecretManagerInRM;
 import org.apache.hadoop.yarn.server.resourcemanager.security.DelegationTokenRenewer;
 import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.RMWebApp;
@@ -146,6 +147,7 @@ public class ResourceManager extends CompositeService implements Recoverable {
   private EventHandler<SchedulerEvent> schedulerDispatcher;
   protected RMAppManager rmAppManager;
   protected ApplicationACLsManager applicationACLsManager;
+  protected QueueACLsManager queueACLsManager;
   protected RMDelegationTokenSecretManager rmDTSecretManager;
   private DelegationTokenRenewer delegationTokenRenewer;
   private WebApp webApp;
@@ -184,6 +186,11 @@ public class ResourceManager extends CompositeService implements Recoverable {
     super.serviceInit(conf);
   }
   
+  protected QueueACLsManager createQueueACLsManager(ResourceScheduler scheduler,
+      Configuration conf) {
+    return new QueueACLsManager(scheduler, conf);
+  }
+
   @VisibleForTesting
   protected void setRMStateStore(RMStateStore rmStore) {
     rmStore.setRMDispatcher(rmDispatcher);
@@ -390,6 +397,8 @@ public class ResourceManager extends CompositeService implements Recoverable {
 
       applicationACLsManager = new ApplicationACLsManager(conf);
 
+      queueACLsManager = createQueueACLsManager(scheduler, conf);
+
       rmAppManager = createRMAppManager();
       // Register event handler for RMAppManagerEvents
       rmDispatcher.register(RMAppManagerEventType.class, rmAppManager);
@@ -803,7 +812,8 @@ public class ResourceManager extends CompositeService implements Recoverable {
 
   protected ClientRMService createClientRMService() {
     return new ClientRMService(this.rmContext, scheduler, this.rmAppManager,
-        this.applicationACLsManager, this.rmDTSecretManager);
+        this.applicationACLsManager, this.queueACLsManager,
+        this.rmDTSecretManager);
   }
 
   protected ApplicationMasterService createApplicationMasterService() {
@@ -883,6 +893,11 @@ public class ResourceManager extends CompositeService implements Recoverable {
     return this.applicationACLsManager;
   }
 
+  @Private
+  public QueueACLsManager getQueueACLsManager() {
+    return this.queueACLsManager;
+  }
+
   @Private
   public RMContainerTokenSecretManager getRMContainerTokenSecretManager() {
     return this.containerTokenSecretManager;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/Queue.java

@@ -64,4 +64,6 @@ public interface Queue {
    * @return queue ACLs for user
    */
   List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user);
+
+  boolean hasAccess(QueueACL acl, UserGroupInformation user);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/YarnScheduler.java

@@ -25,9 +25,11 @@ import org.apache.hadoop.classification.InterfaceAudience.LimitedPrivate;
 import org.apache.hadoop.classification.InterfaceAudience.Public;
 import org.apache.hadoop.classification.InterfaceStability.Evolving;
 import org.apache.hadoop.classification.InterfaceStability.Stable;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
@@ -134,4 +136,17 @@ public interface YarnScheduler extends EventHandler<SchedulerEvent> {
   @LimitedPrivate("yarn")
   @Evolving
   QueueMetrics getRootQueueMetrics();
+
+  /**
+   * Check if the user has permission to perform the operation.
+   * If the user has {@link QueueACL#ADMINISTER_QUEUE} permission,
+   * this user can view/modify the applications in this queue
+   * @param callerUGI
+   * @param acl
+   * @param queueName
+   * @return <code>true</code> if the user has the permission,
+   *         <code>false</code> otherwise
+   */
+  boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName);
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/CapacityScheduler.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.api.records.Container;
 import org.apache.hadoop.yarn.api.records.ContainerId;
 import org.apache.hadoop.yarn.api.records.ContainerStatus;
 import org.apache.hadoop.yarn.api.records.NodeId;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.QueueUserACLInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
@@ -913,4 +914,18 @@ public class CapacityScheduler
         RMContainerEventType.KILL);
   }
 
+  @Override
+  public synchronized boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName) {
+    CSQueue queue = getQueue(queueName);
+    if (queue == null) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("ACL not found for queue access-type " + acl
+            + " for queue " + queueName);
+      }
+      return false;
+    }
+    return queue.hasAccess(acl, callerUGI);
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/LeafQueue.java

@@ -644,7 +644,8 @@ public class LeafQueue implements CSQueue {
 
     // Check queue ACLs
     UserGroupInformation userUgi = UserGroupInformation.createRemoteUser(userName);
-    if (!hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi)) {
+    if (!hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi)
+        && !hasAccess(QueueACL.ADMINISTER_QUEUE, userUgi)) {
       throw new AccessControlException("User " + userName + " cannot submit" +
           " applications to queue " + getQueuePath());
     }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/FairScheduler.java

@@ -624,7 +624,9 @@ public class FairScheduler implements ResourceScheduler {
 
     // Enforce ACLs
     UserGroupInformation userUgi = UserGroupInformation.createRemoteUser(user);
-    if (!queue.hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi)) {
+
+    if (!queue.hasAccess(QueueACL.SUBMIT_APPLICATIONS, userUgi)
+        && !queue.hasAccess(QueueACL.ADMINISTER_QUEUE, userUgi)) {
       String msg = "User " + userUgi.getUserName() +
     	        " cannot submit applications to queue " + queue.getName();
       LOG.info(msg);
@@ -1122,4 +1124,18 @@ public class FairScheduler implements ResourceScheduler {
     return nodes.size();
   }
 
+  @Override
+  public synchronized boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName) {
+    FSQueue queue = getQueueManager().getQueue(queueName);
+    if (queue == null) {
+      if (LOG.isDebugEnabled()) {
+        LOG.debug("ACL not found for queue access-type " + acl
+            + " for queue " + queueName);
+      }
+      return false;
+    }
+    return queue.hasAccess(acl, callerUGI);
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fifo/FifoScheduler.java

@@ -175,6 +175,11 @@ public class FifoScheduler implements ResourceScheduler, Configurable {
       queueUserAclInfo.setUserAcls(Arrays.asList(QueueACL.values()));
       return Collections.singletonList(queueUserAclInfo);
     }
+
+    @Override
+    public boolean hasAccess(QueueACL acl, UserGroupInformation user) {
+      return getQueueAcls().get(acl).isUserAllowed(user);
+    }
   };
 
   @Override
@@ -836,4 +841,10 @@ public class FifoScheduler implements ResourceScheduler, Configurable {
     return DEFAULT_QUEUE.getMetrics();
   }
 
+  @Override
+  public synchronized boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName) {
+    return DEFAULT_QUEUE.hasAccess(acl, callerUGI);
+  }
+
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/security/QueueACLsManager.java

@@ -0,0 +1,44 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.hadoop.yarn.server.resourcemanager.security;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.api.records.QueueACL;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+
+public class QueueACLsManager {
+  private ResourceScheduler scheduler;
+  private boolean isACLsEnable;
+
+  public QueueACLsManager(ResourceScheduler scheduler, Configuration conf) {
+    this.scheduler = scheduler;
+    this.isACLsEnable = conf.getBoolean(YarnConfiguration.YARN_ACL_ENABLE,
+        YarnConfiguration.DEFAULT_YARN_ACL_ENABLE);
+  }
+
+  public boolean checkAccess(UserGroupInformation callerUGI,
+      QueueACL acl, String queueName) {
+    if (!isACLsEnable) {
+      return true;
+    }
+    return scheduler.checkAccess(callerUGI, acl, queueName);
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/AppBlock.java

@@ -25,22 +25,21 @@ import static org.apache.hadoop.yarn.webapp.view.JQueryUI._INFO_WRAP;
 import static org.apache.hadoop.yarn.webapp.view.JQueryUI._ODD;
 import static org.apache.hadoop.yarn.webapp.view.JQueryUI._TH;
 
-
 import java.util.Collection;
 
-import com.google.inject.Inject;
-
 import org.apache.hadoop.http.HttpConfig;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
+import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppAttemptInfo;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppInfo;
-import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.util.Apps;
 import org.apache.hadoop.yarn.util.Times;
@@ -50,14 +49,19 @@ import org.apache.hadoop.yarn.webapp.hamlet.Hamlet.TABLE;
 import org.apache.hadoop.yarn.webapp.view.HtmlBlock;
 import org.apache.hadoop.yarn.webapp.view.InfoBlock;
 
+import com.google.inject.Inject;
+
 public class AppBlock extends HtmlBlock {
 
   private ApplicationACLsManager aclsManager;
+  private QueueACLsManager queueACLsManager;
 
   @Inject
-  AppBlock(ResourceManager rm, ViewContext ctx, ApplicationACLsManager aclsManager) {
+  AppBlock(ResourceManager rm, ViewContext ctx,
+      ApplicationACLsManager aclsManager, QueueACLsManager queueACLsManager) {
     super(ctx);
     this.aclsManager = aclsManager;
+    this.queueACLsManager = queueACLsManager;
   }
 
   @Override
@@ -91,8 +95,10 @@ public class AppBlock extends HtmlBlock {
       callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
     }
     if (callerUGI != null
-        && !this.aclsManager.checkAccess(callerUGI,
+        && !(this.aclsManager.checkAccess(callerUGI,
-            ApplicationAccessType.VIEW_APP, app.getUser(), appID)) {
+                ApplicationAccessType.VIEW_APP, app.getUser(), appID) ||
+             this.queueACLsManager.checkAccess(callerUGI,
+                QueueACL.ADMINISTER_QUEUE, app.getQueue()))) {
       puts("You (User " + remoteUser
           + ") are not authorized to view application " + appID);
       return;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebApp.java

@@ -22,6 +22,7 @@ import static org.apache.hadoop.yarn.util.StringHelper.pajoin;
 
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.apache.hadoop.yarn.webapp.WebApp;
@@ -48,6 +49,7 @@ public class RMWebApp extends WebApp implements YarnWebParams {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
     }
     route("/", RmController.class);
     route(pajoin("/nodes", NODE_STATE), RmController.class, "nodes");

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/RMWebServices.java

@@ -44,6 +44,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.FinalApplicationStatus;
 import org.apache.hadoop.yarn.api.records.NodeId;
 import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.YarnApplicationState;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
@@ -57,6 +58,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppAttemptInfo;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppAttemptsInfo;
 import org.apache.hadoop.yarn.server.resourcemanager.webapp.dao.AppInfo;
@@ -89,14 +91,17 @@ public class RMWebServices {
   private static RecordFactory recordFactory = RecordFactoryProvider
       .getRecordFactory(null);
   private final ApplicationACLsManager aclsManager;
+  private final QueueACLsManager queueACLsManager;
 
   private @Context HttpServletResponse response;
 
   @Inject
   public RMWebServices(final ResourceManager rm,
-      final ApplicationACLsManager aclsManager) {
+      final ApplicationACLsManager aclsManager,
+      final QueueACLsManager queueACLsManager) {
     this.rm = rm;
     this.aclsManager = aclsManager;
+    this.queueACLsManager = queueACLsManager;
   }
 
   protected Boolean hasAccess(RMApp app, HttpServletRequest hsr) {
@@ -107,9 +112,10 @@ public class RMWebServices {
       callerUGI = UserGroupInformation.createRemoteUser(remoteUser);
     }
     if (callerUGI != null
-        && !this.aclsManager.checkAccess(callerUGI,
+        && !(this.aclsManager.checkAccess(callerUGI,
             ApplicationAccessType.VIEW_APP, app.getUser(),
-            app.getApplicationId())) {
+            app.getApplicationId()) || this.queueACLsManager.checkAccess(
+            callerUGI, QueueACL.ADMINISTER_QUEUE, app.getQueue()))) {
       return false;
     }
     return true;

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/MockRM.java

@@ -299,7 +299,8 @@ public class MockRM extends ResourceManager {
   @Override
   protected ClientRMService createClientRMService() {
     return new ClientRMService(getRMContext(), getResourceScheduler(),
-        rmAppManager, applicationACLsManager, rmDTSecretManager) {
+        rmAppManager, applicationACLsManager, queueACLsManager,
+        rmDTSecretManager) {
       @Override
       protected void serviceStart() {
         // override to not start rpc handler

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestApplicationACLs.java

@@ -18,6 +18,10 @@
 
 package org.apache.hadoop.yarn.server.resourcemanager;
 
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
 import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.security.PrivilegedExceptionAction;
@@ -45,6 +49,7 @@ import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationResourceUsageReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
 import org.apache.hadoop.yarn.exceptions.YarnException;
@@ -54,16 +59,21 @@ import org.apache.hadoop.yarn.ipc.YarnRPC;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStoreFactory;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 public class TestApplicationACLs {
 
   private static final String APP_OWNER = "owner";
   private static final String FRIEND = "friend";
   private static final String ENEMY = "enemy";
+  private static final String QUEUE_ADMIN_USER = "queue-admin-user";
   private static final String SUPER_USER = "superUser";
   private static final String FRIENDLY_GROUP = "friendly-group";
   private static final String SUPER_GROUP = "superGroup";
@@ -83,6 +93,8 @@ public class TestApplicationACLs {
   private static RecordFactory recordFactory = RecordFactoryProvider
       .getRecordFactory(conf);
 
+  private static boolean isQueueUser = false;
+
   @BeforeClass
   public static void setup() throws InterruptedException, IOException {
     RMStateStore store = RMStateStoreFactory.getStore(conf);
@@ -91,9 +103,25 @@ public class TestApplicationACLs {
     adminACL.addGroup(SUPER_GROUP);
     conf.set(YarnConfiguration.YARN_ADMIN_ACL, adminACL.getAclString());
     resourceManager = new MockRM(conf) {
+
+      @Override
+      protected QueueACLsManager createQueueACLsManager(
+          ResourceScheduler scheduler,
+          Configuration conf) {
+        QueueACLsManager mockQueueACLsManager = mock(QueueACLsManager.class);
+        when(mockQueueACLsManager.checkAccess(any(UserGroupInformation.class),
+            any(QueueACL.class), anyString())).thenAnswer(new Answer() {
+          public Object answer(InvocationOnMock invocation) {
+            return isQueueUser;
+          }
+        });
+        return mockQueueACLsManager;
+      }
+
       protected ClientRMService createClientRMService() {
         return new ClientRMService(getRMContext(), this.scheduler,
-            this.rmAppManager, this.applicationACLsManager, null);
+            this.rmAppManager, this.applicationACLsManager,
+            this.queueACLsManager, null);
       };
     };
     new Thread() {
@@ -147,6 +175,8 @@ public class TestApplicationACLs {
     verifyFriendAccess();
 
     verifyEnemyAccess();
+
+    verifyAdministerQueueUserAccess();
   }
 
   private ApplicationId submitAppAndGetAppId(AccessControlList viewACL,
@@ -358,4 +388,36 @@ public class TestApplicationACLs {
     Assert.assertEquals("Enemy should not see app needed resources",
         -1, usageReport.getNeededResources().getMemory());
   }
+
+  private void verifyAdministerQueueUserAccess() throws Exception {
+    isQueueUser = true;
+    AccessControlList viewACL = new AccessControlList("");
+    viewACL.addGroup(FRIENDLY_GROUP);
+    AccessControlList modifyACL = new AccessControlList("");
+    modifyACL.addUser(FRIEND);
+    ApplicationId applicationId = submitAppAndGetAppId(viewACL, modifyACL);
+
+    final GetApplicationReportRequest appReportRequest = recordFactory
+        .newRecordInstance(GetApplicationReportRequest.class);
+    appReportRequest.setApplicationId(applicationId);
+    final KillApplicationRequest finishAppRequest = recordFactory
+        .newRecordInstance(KillApplicationRequest.class);
+    finishAppRequest.setApplicationId(applicationId);
+
+    ApplicationClientProtocol administerQueueUserRmClient =
+        getRMClientForUser(QUEUE_ADMIN_USER);
+
+    // View as the administerQueueUserRmClient
+    administerQueueUserRmClient.getApplicationReport(appReportRequest);
+
+    // List apps as administerQueueUserRmClient
+    Assert.assertEquals("App view by queue-admin-user should list the apps!!",
+        5, administerQueueUserRmClient.getApplications(
+               recordFactory.newRecordInstance(GetApplicationsRequest.class))
+               .getApplicationList().size());
+
+    // Kill app as the administerQueueUserRmClient
+    administerQueueUserRmClient.forceKillApplication(finishAppRequest);
+    resourceManager.waitForState(applicationId, RMAppState.KILLED);
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java

@@ -19,9 +19,12 @@
 package org.apache.hadoop.yarn.server.resourcemanager;
 
 import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
+import static org.mockito.Mockito.spy;
 
 import java.io.IOException;
 import java.net.InetSocketAddress;
@@ -55,12 +58,14 @@ import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.RenewDelegationTokenRequest;
 import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
 import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
+import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
 import org.apache.hadoop.yarn.api.records.ApplicationId;
 import org.apache.hadoop.yarn.api.records.ApplicationReport;
 import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
 import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
 import org.apache.hadoop.yarn.api.records.NodeReport;
 import org.apache.hadoop.yarn.api.records.NodeState;
+import org.apache.hadoop.yarn.api.records.QueueACL;
 import org.apache.hadoop.yarn.api.records.QueueInfo;
 import org.apache.hadoop.yarn.api.records.Resource;
 import org.apache.hadoop.yarn.conf.YarnConfiguration;
@@ -78,7 +83,9 @@ import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppImpl;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
@@ -119,7 +126,7 @@ public class TestClientRMService {
     MockRM rm = new MockRM() {
       protected ClientRMService createClientRMService() {
         return new ClientRMService(this.rmContext, scheduler,
-          this.rmAppManager, this.applicationACLsManager,
+          this.rmAppManager, this.applicationACLsManager, this.queueACLsManager,
           this.rmDTSecretManager);
       };
     };
@@ -183,7 +190,7 @@ public class TestClientRMService {
     when(rmContext.getRMApps()).thenReturn(
         new ConcurrentHashMap<ApplicationId, RMApp>());
     ClientRMService rmService = new ClientRMService(rmContext, null, null,
-        null, null);
+        null, null, null);
     RecordFactory recordFactory = RecordFactoryProvider.getRecordFactory(null);
     GetApplicationReportRequest request = recordFactory
         .newRecordInstance(GetApplicationReportRequest.class);
@@ -204,7 +211,7 @@ public class TestClientRMService {
     when(rmContext.getRMApps()).thenReturn(
         new ConcurrentHashMap<ApplicationId, RMApp>());
     ClientRMService rmService = new ClientRMService(rmContext, null, null,
-        null, null);
+        null, null, null);
     ApplicationId applicationId =
         BuilderUtils.newApplicationId(System.currentTimeMillis(), 0);
     KillApplicationRequest request =
@@ -225,7 +232,7 @@ public class TestClientRMService {
     RMContext rmContext = mock(RMContext.class);
     mockRMContext(yarnScheduler, rmContext);
     ClientRMService rmService = new ClientRMService(rmContext, yarnScheduler,
-        null, null, null);
+        null, null, null, null);
     GetQueueInfoRequest request = recordFactory
         .newRecordInstance(GetQueueInfoRequest.class);
     request.setQueueName("testqueue");
@@ -308,7 +315,7 @@ public class TestClientRMService {
 
     RMContext rmContext = mock(RMContext.class);
     ClientRMService rmService = new ClientRMService(
-        rmContext, null, null, null, dtsm);
+        rmContext, null, null, null, null, dtsm);
     rmService.renewDelegationToken(request);
   }
 
@@ -332,9 +339,13 @@ public class TestClientRMService {
     when(
         mockAclsManager.checkAccess(UserGroupInformation.getCurrentUser(),
             ApplicationAccessType.VIEW_APP, null, appId1)).thenReturn(true);
+
+    QueueACLsManager mockQueueACLsManager = mock(QueueACLsManager.class);
+    when(mockQueueACLsManager.checkAccess(any(UserGroupInformation.class),
+            any(QueueACL.class), anyString())).thenReturn(true);
     ClientRMService rmService =
         new ClientRMService(rmContext, yarnScheduler, appManager,
-            mockAclsManager, null);
+            mockAclsManager, mockQueueACLsManager, null);
 
     // without name and queue
 
@@ -444,7 +455,8 @@ public class TestClientRMService {
     when(rmContext.getDispatcher().getEventHandler()).thenReturn(eventHandler);
       
     final ClientRMService rmService =
-        new ClientRMService(rmContext, yarnScheduler, appManager, null, null);
+        new ClientRMService(rmContext, yarnScheduler, appManager, null, null,
+            null);
 
     // submit an app and wait for it to block while in app submission
     Thread t = new Thread() {
@@ -490,6 +502,8 @@ public class TestClientRMService {
       throws IOException {
     Dispatcher dispatcher = mock(Dispatcher.class);
     when(rmContext.getDispatcher()).thenReturn(dispatcher);
+    EventHandler eventHandler = mock(EventHandler.class);
+    when(dispatcher.getEventHandler()).thenReturn(eventHandler);
     QueueInfo queInfo = recordFactory.newRecordInstance(QueueInfo.class);
     queInfo.setQueueName("testqueue");
     when(yarnScheduler.getQueueInfo(eq("testqueue"), anyBoolean(), anyBoolean()))
@@ -526,9 +540,14 @@ public class TestClientRMService {
       ApplicationId applicationId3, YarnConfiguration config, String queueName) {
     ApplicationSubmissionContext asContext = mock(ApplicationSubmissionContext.class);
     when(asContext.getMaxAppAttempts()).thenReturn(1);
-    return new RMAppImpl(applicationId3, rmContext, config, null, null,
+    RMAppImpl app =  spy(new RMAppImpl(applicationId3, rmContext, config, null, null,
         queueName, asContext, yarnScheduler, null , System
-            .currentTimeMillis(), "YARN");
+            .currentTimeMillis(), "YARN"));
+    ApplicationAttemptId attemptId = ApplicationAttemptId.newInstance(applicationId3, 1);
+    RMAppAttemptImpl rmAppAttemptImpl = new RMAppAttemptImpl(attemptId,
+        rmContext, yarnScheduler, null, asContext, config, null);
+    when(app.getCurrentAppAttempt()).thenReturn(rmAppAttemptImpl);
+    return app;
   }
 
   private static YarnScheduler mockYarnScheduler() {

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMTokens.java

@@ -63,6 +63,7 @@ import org.apache.hadoop.yarn.ipc.YarnRPC;
 import org.apache.hadoop.yarn.security.client.RMDelegationTokenIdentifier;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.NullRMStateStore;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.resourcemanager.security.RMDelegationTokenSecretManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
@@ -429,7 +430,8 @@ public class TestClientRMTokens {
         ResourceScheduler scheduler,
         RMDelegationTokenSecretManager rmDTSecretManager) {
       super(mock(RMContext.class), scheduler, mock(RMAppManager.class),
-          new ApplicationACLsManager(conf), rmDTSecretManager);
+          new ApplicationACLsManager(conf), new QueueACLsManager(scheduler,
+              conf), rmDTSecretManager);
     }
 
     // Use a random port unless explicitly specified.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestQueueACLs.java

@@ -0,0 +1,306 @@
+/**
+* Licensed to the Apache Software Foundation (ASF) under one
+* or more contributor license agreements.  See the NOTICE file
+* distributed with this work for additional information
+* regarding copyright ownership.  The ASF licenses this file
+* to you under the Apache License, Version 2.0 (the
+* "License"); you may not use this file except in compliance
+* with the License.  You may obtain a copy of the License at
+*
+*     http://www.apache.org/licenses/LICENSE-2.0
+*
+* Unless required by applicable law or agreed to in writing, software
+* distributed under the License is distributed on an "AS IS" BASIS,
+* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+* See the License for the specific language governing permissions and
+* limitations under the License.
+*/
+
+package org.apache.hadoop.yarn.server.resourcemanager;
+
+import java.io.IOException;
+import java.net.InetSocketAddress;
+import java.security.PrivilegedExceptionAction;
+import java.util.HashMap;
+import java.util.Map;
+
+import junit.framework.Assert;
+
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.service.Service.STATE;
+import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationReportResponse;
+import org.apache.hadoop.yarn.api.protocolrecords.GetNewApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.KillApplicationRequest;
+import org.apache.hadoop.yarn.api.protocolrecords.SubmitApplicationRequest;
+import org.apache.hadoop.yarn.api.records.ApplicationAccessType;
+import org.apache.hadoop.yarn.api.records.ApplicationId;
+import org.apache.hadoop.yarn.api.records.ApplicationSubmissionContext;
+import org.apache.hadoop.yarn.api.records.ContainerLaunchContext;
+import org.apache.hadoop.yarn.api.records.QueueACL;
+import org.apache.hadoop.yarn.api.records.Resource;
+import org.apache.hadoop.yarn.conf.YarnConfiguration;
+import org.apache.hadoop.yarn.exceptions.YarnException;
+import org.apache.hadoop.yarn.ipc.YarnRPC;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
+import org.junit.AfterClass;
+import org.junit.BeforeClass;
+import org.junit.Test;
+
+public class TestQueueACLs {
+
+  private static final String COMMON_USER = "common_user";
+  private static final String QUEUE_A_USER = "queueA_user";
+  private static final String QUEUE_B_USER = "queueB_user";
+  private static final String ROOT_ADMIN = "root_admin";
+  private static final String QUEUE_A_ADMIN = "queueA_admin";
+  private static final String QUEUE_B_ADMIN = "queueB_admin";
+
+  private static final String QUEUEA = "queueA";
+  private static final String QUEUEB = "queueB";
+
+  private static final Log LOG = LogFactory.getLog(TestApplicationACLs.class);
+
+  static MockRM resourceManager;
+  static Configuration conf = createConfiguration();
+  final static YarnRPC rpc = YarnRPC.create(conf);
+  final static InetSocketAddress rmAddress = conf.getSocketAddr(
+    YarnConfiguration.RM_ADDRESS, YarnConfiguration.DEFAULT_RM_ADDRESS,
+    YarnConfiguration.DEFAULT_RM_PORT);
+
+  @BeforeClass
+  public static void setup() throws InterruptedException, IOException {
+    AccessControlList adminACL = new AccessControlList("");
+    conf.set(YarnConfiguration.YARN_ADMIN_ACL, adminACL.getAclString());
+
+    resourceManager = new MockRM(conf) {
+      protected ClientRMService createClientRMService() {
+        return new ClientRMService(getRMContext(), this.scheduler,
+          this.rmAppManager, this.applicationACLsManager,
+          this.queueACLsManager, this.rmDTSecretManager);
+      };
+
+      @Override
+      protected void doSecureLogin() throws IOException {
+      }
+    };
+    new Thread() {
+      public void run() {
+        resourceManager.start();
+      };
+    }.start();
+    int waitCount = 0;
+    while (resourceManager.getServiceState() == STATE.INITED
+        && waitCount++ < 60) {
+      LOG.info("Waiting for RM to start...");
+      Thread.sleep(1500);
+    }
+    if (resourceManager.getServiceState() != STATE.STARTED) {
+      // RM could have failed.
+      throw new IOException("ResourceManager failed to start. Final state is "
+          + resourceManager.getServiceState());
+    }
+  }
+
+  @AfterClass
+  public static void tearDown() {
+    if (resourceManager != null) {
+      resourceManager.stop();
+    }
+  }
+
+  @Test
+  public void testApplicationACLs() throws Exception {
+
+    verifyKillAppSuccess(QUEUE_A_USER, QUEUE_A_USER, QUEUEA, true);
+    verifyKillAppSuccess(QUEUE_A_USER, QUEUE_A_ADMIN, QUEUEA, true);
+    verifyKillAppSuccess(QUEUE_A_USER, COMMON_USER, QUEUEA, true);
+    verifyKillAppSuccess(QUEUE_A_USER, ROOT_ADMIN, QUEUEA, true);
+    verifyKillAppFailure(QUEUE_A_USER, QUEUE_B_USER, QUEUEA, true);
+    verifyKillAppFailure(QUEUE_A_USER, QUEUE_B_ADMIN, QUEUEA, true);
+
+    verifyKillAppSuccess(QUEUE_B_USER, QUEUE_B_USER, QUEUEB, true);
+    verifyKillAppSuccess(QUEUE_B_USER, QUEUE_B_ADMIN, QUEUEB, true);
+    verifyKillAppSuccess(QUEUE_B_USER, COMMON_USER, QUEUEB, true);
+    verifyKillAppSuccess(QUEUE_B_USER, ROOT_ADMIN, QUEUEB, true);
+
+    verifyKillAppFailure(QUEUE_B_USER, QUEUE_A_USER, QUEUEB, true);
+    verifyKillAppFailure(QUEUE_B_USER, QUEUE_A_ADMIN, QUEUEB, true);
+
+    verifyKillAppSuccess(ROOT_ADMIN, ROOT_ADMIN, QUEUEA, false);
+    verifyKillAppSuccess(ROOT_ADMIN, ROOT_ADMIN, QUEUEB, false);
+
+    verifyGetClientAMToken(QUEUE_A_USER, ROOT_ADMIN, QUEUEA, true);
+
+  }
+
+  private void verifyGetClientAMToken(String submitter, String queueAdmin,
+      String queueName, boolean setupACLs) throws Exception {
+    ApplicationId applicationId =
+        submitAppAndGetAppId(submitter, queueName, setupACLs);
+    final GetApplicationReportRequest appReportRequest =
+        GetApplicationReportRequest.newInstance(applicationId);
+
+    ApplicationClientProtocol submitterClient = getRMClientForUser(submitter);
+    ApplicationClientProtocol adMinUserClient = getRMClientForUser(queueAdmin);
+
+    GetApplicationReportResponse submitterGetReport =
+        submitterClient.getApplicationReport(appReportRequest);
+    GetApplicationReportResponse adMinUserGetReport =
+        adMinUserClient.getApplicationReport(appReportRequest);
+
+    Assert.assertEquals(submitterGetReport.getApplicationReport()
+      .getClientToAMToken(), adMinUserGetReport.getApplicationReport()
+      .getClientToAMToken());
+  }
+
+  private void verifyKillAppFailure(String submitter, String killer,
+      String queueName, boolean setupACLs) throws Exception {
+
+    ApplicationId applicationId =
+        submitAppAndGetAppId(submitter, queueName, setupACLs);
+
+    final KillApplicationRequest finishAppRequest =
+        KillApplicationRequest.newInstance(applicationId);
+
+    ApplicationClientProtocol killerClient = getRMClientForUser(killer);
+
+    // Kill app as the killer
+    try {
+      killerClient.forceKillApplication(finishAppRequest);
+      Assert.fail("App killing by the enemy should fail!!");
+    } catch (YarnException e) {
+      LOG.info("Got exception while killing app as the enemy", e);
+      Assert.assertTrue(e.getMessage().contains(
+        "User " + killer + " cannot perform operation MODIFY_APP on "
+            + applicationId));
+    }
+
+    getRMClientForUser(submitter).forceKillApplication(finishAppRequest);
+  }
+
+  private void verifyKillAppSuccess(String submitter, String killer,
+      String queueName, boolean setupACLs) throws Exception {
+    ApplicationId applicationId =
+        submitAppAndGetAppId(submitter, queueName, setupACLs);
+
+    final KillApplicationRequest finishAppRequest =
+        KillApplicationRequest.newInstance(applicationId);
+
+    ApplicationClientProtocol ownerClient = getRMClientForUser(killer);
+
+    // Kill app as killer
+    ownerClient.forceKillApplication(finishAppRequest);
+    resourceManager.waitForState(applicationId, RMAppState.KILLED);
+  }
+
+  private ApplicationId submitAppAndGetAppId(String submitter,
+      String queueName, boolean setupACLs) throws Exception {
+
+    GetNewApplicationRequest newAppRequest =
+        GetNewApplicationRequest.newInstance();
+
+    ApplicationClientProtocol submitterClient = getRMClientForUser(submitter);
+    ApplicationId applicationId =
+        submitterClient.getNewApplication(newAppRequest).getApplicationId();
+
+    Resource resource = BuilderUtils.newResource(1024, 1);
+    Map<ApplicationAccessType, String> acls = createACLs(submitter, setupACLs);
+    ContainerLaunchContext amContainerSpec =
+        ContainerLaunchContext.newInstance(null, null, null, null, null, acls);
+
+    ApplicationSubmissionContext appSubmissionContext =
+        ApplicationSubmissionContext.newInstance(applicationId,
+          "applicationName", queueName, null, amContainerSpec, false, true, 1,
+          resource, "applicationType");
+    appSubmissionContext.setApplicationId(applicationId);
+    appSubmissionContext.setQueue(queueName);
+
+    SubmitApplicationRequest submitRequest =
+        SubmitApplicationRequest.newInstance(appSubmissionContext);
+    submitterClient.submitApplication(submitRequest);
+    resourceManager.waitForState(applicationId, RMAppState.ACCEPTED);
+    return applicationId;
+  }
+
+  private Map<ApplicationAccessType, String> createACLs(String submitter,
+      boolean setupACLs) {
+    AccessControlList viewACL = new AccessControlList("");
+    AccessControlList modifyACL = new AccessControlList("");
+    if (setupACLs) {
+      viewACL.addUser(submitter);
+      viewACL.addUser(COMMON_USER);
+      modifyACL.addUser(submitter);
+      modifyACL.addUser(COMMON_USER);
+    }
+    Map<ApplicationAccessType, String> acls =
+        new HashMap<ApplicationAccessType, String>();
+    acls.put(ApplicationAccessType.VIEW_APP, viewACL.getAclString());
+    acls.put(ApplicationAccessType.MODIFY_APP, modifyACL.getAclString());
+    return acls;
+  }
+
+  private ApplicationClientProtocol getRMClientForUser(String user)
+      throws IOException, InterruptedException {
+    UserGroupInformation userUGI = UserGroupInformation.createRemoteUser(user);
+    ApplicationClientProtocol userClient =
+        userUGI
+          .doAs(new PrivilegedExceptionAction<ApplicationClientProtocol>() {
+            @Override
+            public ApplicationClientProtocol run() throws Exception {
+              return (ApplicationClientProtocol) rpc.getProxy(
+                ApplicationClientProtocol.class, rmAddress, conf);
+            }
+          });
+    return userClient;
+  }
+
+  private static YarnConfiguration createConfiguration() {
+    CapacitySchedulerConfiguration csConf =
+        new CapacitySchedulerConfiguration();
+    csConf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {
+        QUEUEA, QUEUEB });
+
+    csConf.setCapacity(CapacitySchedulerConfiguration.ROOT + "." + QUEUEA, 50f);
+    csConf.setCapacity(CapacitySchedulerConfiguration.ROOT + "." + QUEUEB, 50f);
+
+    Map<QueueACL, AccessControlList> aclsOnQueueA =
+        new HashMap<QueueACL, AccessControlList>();
+    AccessControlList submitACLonQueueA = new AccessControlList(QUEUE_A_USER);
+    submitACLonQueueA.addUser(COMMON_USER);
+    AccessControlList adminACLonQueueA = new AccessControlList(QUEUE_A_ADMIN);
+    aclsOnQueueA.put(QueueACL.SUBMIT_APPLICATIONS, submitACLonQueueA);
+    aclsOnQueueA.put(QueueACL.ADMINISTER_QUEUE, adminACLonQueueA);
+    csConf.setAcls(CapacitySchedulerConfiguration.ROOT + "." + QUEUEA,
+      aclsOnQueueA);
+
+    Map<QueueACL, AccessControlList> aclsOnQueueB =
+        new HashMap<QueueACL, AccessControlList>();
+    AccessControlList submitACLonQueueB = new AccessControlList(QUEUE_B_USER);
+    submitACLonQueueB.addUser(COMMON_USER);
+    AccessControlList adminACLonQueueB = new AccessControlList(QUEUE_B_ADMIN);
+    aclsOnQueueB.put(QueueACL.SUBMIT_APPLICATIONS, submitACLonQueueB);
+    aclsOnQueueB.put(QueueACL.ADMINISTER_QUEUE, adminACLonQueueB);
+    csConf.setAcls(CapacitySchedulerConfiguration.ROOT + "." + QUEUEB,
+      aclsOnQueueB);
+
+    Map<QueueACL, AccessControlList> aclsOnRootQueue =
+        new HashMap<QueueACL, AccessControlList>();
+    AccessControlList submitACLonRoot = new AccessControlList("");
+    AccessControlList adminACLonRoot = new AccessControlList(ROOT_ADMIN);
+    aclsOnRootQueue.put(QueueACL.SUBMIT_APPLICATIONS, submitACLonRoot);
+    aclsOnRootQueue.put(QueueACL.ADMINISTER_QUEUE, adminACLonRoot);
+    csConf.setAcls(CapacitySchedulerConfiguration.ROOT, aclsOnRootQueue);
+
+    YarnConfiguration conf = new YarnConfiguration(csConf);
+    conf.setBoolean(YarnConfiguration.YARN_ACL_ENABLE, true);
+    return conf;
+  }
+}

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestRMRestart.java

@@ -749,7 +749,7 @@ public class TestRMRestart {
     @Override
     protected ClientRMService createClientRMService() {
       return new ClientRMService(getRMContext(), getResourceScheduler(),
-          rmAppManager, applicationACLsManager, rmDTSecretManager){ 
+          rmAppManager, applicationACLsManager, null, rmDTSecretManager){
         @Override
         protected void serviceStart() throws Exception {
           // do nothing

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/capacity/TestLeafQueue.java

@@ -151,7 +151,8 @@ public class TestLeafQueue {
     // Define top-level queues
     conf.setQueues(CapacitySchedulerConfiguration.ROOT, new String[] {newRoot});
     conf.setMaximumCapacity(CapacitySchedulerConfiguration.ROOT, 100);
-    conf.setAcl(CapacitySchedulerConfiguration.ROOT, QueueACL.SUBMIT_APPLICATIONS, " ");
+    conf.setAcl(CapacitySchedulerConfiguration.ROOT,
+      QueueACL.SUBMIT_APPLICATIONS, " ");
     
     final String Q_newRoot = CapacitySchedulerConfiguration.ROOT + "." + newRoot;
     conf.setQueues(Q_newRoot, new String[] {A, B, C, D, E});

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/scheduler/fair/TestFairScheduler.java

@@ -23,6 +23,10 @@ import static org.junit.Assert.assertNotNull;
 import static org.junit.Assert.assertNull;
 import static org.junit.Assert.assertTrue;
 import static org.junit.Assert.fail;
+import static org.mockito.Matchers.any;
+import static org.mockito.Matchers.anyString;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
 
 import java.io.File;
 import java.io.FileWriter;
@@ -32,6 +36,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.Collections;
+import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 
@@ -41,6 +46,7 @@ import junit.framework.Assert;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.metrics2.lib.DefaultMetricsSystem;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.AccessControlList;
 import org.apache.hadoop.yarn.MockApps;
 import org.apache.hadoop.yarn.api.records.ApplicationAttemptId;
@@ -72,6 +78,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmcontainer.RMContainer;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.TestCapacityScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppAddedSchedulerEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.AppRemovedSchedulerEvent;
@@ -81,12 +88,15 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.event.NodeUpdateS
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.DominantResourceFairnessPolicy;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.policies.FifoPolicy;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.apache.hadoop.yarn.util.Clock;
 import org.apache.hadoop.yarn.util.resource.Resources;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 import org.xml.sax.SAXException;
 
 public class TestFairScheduler {
@@ -1538,6 +1548,7 @@ public class TestFairScheduler {
     out.println("<allocations>");
     out.println("<queue name=\"queue1\">");
     out.println("<aclSubmitApps>norealuserhasthisname</aclSubmitApps>");
+    out.println("<aclAdministerApps>norealuserhasthisname</aclAdministerApps>");
     out.println("</queue>");
     out.println("</allocations>");
     out.close();
@@ -1766,6 +1777,7 @@ public class TestFairScheduler {
     out.println("<allocations>");
     out.println("<queue name=\"queue1\">");
     out.println("<aclSubmitApps>userallow</aclSubmitApps>");
+    out.println("<aclAdministerApps>userallow</aclAdministerApps>");
     out.println("</queue>");
     out.println("</allocations>");
     out.close();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/security/TestClientToAMTokens.java

@@ -167,7 +167,7 @@ public class TestClientToAMTokens {
     MockRM rm = new MockRMWithCustomAMLauncher(conf, containerManager) {
       protected ClientRMService createClientRMService() {
         return new ClientRMService(this.rmContext, scheduler,
-          this.rmAppManager, this.applicationACLsManager,
+          this.rmAppManager, this.applicationACLsManager, this.queueACLsManager,
           this.rmDTSecretManager);
       };
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServices.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.QueueMetrics;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.util.YarnVersionInfo;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
@@ -84,6 +85,7 @@ public class TestRMWebServices extends JerseyTest {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
       serve("/*").with(GuiceContainer.class);
     }
   });

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesApps.java

@@ -48,6 +48,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fifo.FifoScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.apache.hadoop.yarn.webapp.WebServicesTestUtils;
@@ -94,6 +95,7 @@ public class TestRMWebServicesApps extends JerseyTest {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
       serve("/*").with(GuiceContainer.class);
     }
   });

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesCapacitySched.java

@@ -35,6 +35,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacitySchedulerConfiguration;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.util.resource.Resources;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
@@ -107,6 +108,7 @@ public class TestRMWebServicesCapacitySched extends JerseyTest {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
       serve("/*").with(GuiceContainer.class);
     }
   });

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesFairScheduler.java

@@ -28,6 +28,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.codehaus.jettison.json.JSONException;
@@ -62,6 +63,7 @@ public class TestRMWebServicesFairScheduler extends JerseyTest {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
       serve("/*").with(GuiceContainer.class);
     }
   });

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/webapp/TestRMWebServicesNodes.java

@@ -43,6 +43,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNodeStatusEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerNodeReport;
+import org.apache.hadoop.yarn.server.resourcemanager.security.QueueACLsManager;
 import org.apache.hadoop.yarn.server.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.webapp.GenericExceptionHandler;
 import org.apache.hadoop.yarn.webapp.WebServicesTestUtils;
@@ -86,6 +87,7 @@ public class TestRMWebServicesNodes extends JerseyTest {
       bind(RMContext.class).toInstance(rm.getRMContext());
       bind(ApplicationACLsManager.class).toInstance(
           rm.getApplicationACLsManager());
+      bind(QueueACLsManager.class).toInstance(rm.getQueueACLsManager());
       serve("/*").with(GuiceContainer.class);
     }
   });