YARN-4764. Application submission fails when submitted queue is not available in scheduler xml. Contributed by Bibin A Chundatt

(cherry picked from commit 3c33158d1c)

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

@@ -17,7 +17,11 @@
  */
 package org.apache.hadoop.yarn.server.resourcemanager;
 
-import com.google.common.annotations.VisibleForTesting;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.LinkedList;
+import java.util.Map;
+
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
@@ -55,14 +59,12 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.YarnScheduler;
+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.security.ApplicationACLsManager;
 import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 
-import java.io.IOException;
-import java.nio.ByteBuffer;
-import java.util.LinkedList;
-import java.util.Map;
+import com.google.common.annotations.VisibleForTesting;
 
 /**
  * This class manages the list of applications for the resource manager. 
@@ -360,22 +362,23 @@ public class RMAppManager implements EventHandler<RMAppManagerEvent>,
     // mapping should be done outside scheduler too like CS.
     // For now, exclude FS for the acl check.
     if (!isRecovery && YarnConfiguration.isAclEnabled(conf)
-        && scheduler instanceof CapacityScheduler &&
-        !authorizer.checkPermission(new AccessRequest(
-            ((CapacityScheduler) scheduler)
-                .getQueue(submissionContext.getQueue()).getPrivilegedEntity(),
-            userUgi, SchedulerUtils.toAccessType(QueueACL.SUBMIT_APPLICATIONS),
-            submissionContext.getApplicationId().toString(),
-            submissionContext.getApplicationName())) &&
-        !authorizer.checkPermission(new AccessRequest(
-            ((CapacityScheduler) scheduler)
-                .getQueue(submissionContext.getQueue()).getPrivilegedEntity(),
-            userUgi, SchedulerUtils.toAccessType(QueueACL.ADMINISTER_QUEUE),
-            submissionContext.getApplicationId().toString(),
-            submissionContext.getApplicationName()))) {
-      throw new AccessControlException(
-          "User " + user + " does not have permission to submit "
-              + applicationId + " to queue " + submissionContext.getQueue());
+        && scheduler instanceof CapacityScheduler) {
+      String queueName = submissionContext.getQueue();
+      String appName = submissionContext.getApplicationName();
+      CSQueue csqueue = ((CapacityScheduler) scheduler).getQueue(queueName);
+      if (null != csqueue
+          && !authorizer.checkPermission(
+              new AccessRequest(csqueue.getPrivilegedEntity(), userUgi,
+                  SchedulerUtils.toAccessType(QueueACL.SUBMIT_APPLICATIONS),
+                  applicationId.toString(), appName))
+          && !authorizer.checkPermission(
+              new AccessRequest(csqueue.getPrivilegedEntity(), userUgi,
+                  SchedulerUtils.toAccessType(QueueACL.ADMINISTER_QUEUE),
+                  applicationId.toString(), appName))) {
+        throw new AccessControlException(
+            "User " + user + " does not have permission to submit "
+                + applicationId + " to queue " + submissionContext.getQueue());
+      }
     }
 
     // Create RMApp

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

@@ -40,6 +40,7 @@ import org.apache.hadoop.service.Service.STATE;
 import org.apache.hadoop.yarn.api.ApplicationClientProtocol;
 import org.apache.hadoop.yarn.api.protocolrecords.GetApplicationsRequest;
 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;
@@ -179,6 +180,8 @@ public class TestApplicationACLs {
     verifyEnemyAccess();
 
     verifyAdministerQueueUserAccess();
+
+    verifyInvalidQueueWithAcl();
   }
 
   @SuppressWarnings("deprecation")
@@ -392,6 +395,39 @@ public class TestApplicationACLs {
         -1, usageReport.getNeededResources().getMemory());
   }
 
+  private void verifyInvalidQueueWithAcl() throws Exception {
+    isQueueUser = true;
+    SubmitApplicationRequest submitRequest =
+        recordFactory.newRecordInstance(SubmitApplicationRequest.class);
+    ApplicationSubmissionContext context =
+        recordFactory.newRecordInstance(ApplicationSubmissionContext.class);
+    ApplicationId applicationId = rmClient
+        .getNewApplication(
+            recordFactory.newRecordInstance(GetNewApplicationRequest.class))
+        .getApplicationId();
+    context.setApplicationId(applicationId);
+    Map<ApplicationAccessType, String> acls =
+        new HashMap<ApplicationAccessType, String>();
+    ContainerLaunchContext amContainer =
+        recordFactory.newRecordInstance(ContainerLaunchContext.class);
+    Resource resource = BuilderUtils.newResource(1024, 1);
+    context.setResource(resource);
+    amContainer.setApplicationACLs(acls);
+    context.setQueue("InvalidQueue");
+    context.setAMContainerSpec(amContainer);
+    submitRequest.setApplicationSubmissionContext(context);
+    rmClient.submitApplication(submitRequest);
+    resourceManager.waitForState(applicationId, RMAppState.FAILED);
+    final GetApplicationReportRequest appReportRequest =
+        recordFactory.newRecordInstance(GetApplicationReportRequest.class);
+    appReportRequest.setApplicationId(applicationId);
+    GetApplicationReportResponse applicationReport =
+        rmClient.getApplicationReport(appReportRequest);
+    ApplicationReport appReport = applicationReport.getApplicationReport();
+    Assert.assertTrue(appReport.getDiagnostics()
+        .contains("submitted by user owner to unknown queue: InvalidQueue"));
+  }
+
   private void verifyAdministerQueueUserAccess() throws Exception {
     isQueueUser = true;
     AccessControlList viewACL = new AccessControlList("");