YARN-333. Schedulers cannot control the queue-name of an application. (sandyr via tucu)

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

hadoop-yarn-project/CHANGES.txt

@@ -710,6 +710,9 @@ Release 2.1.0-beta - 2013-07-02
     YARN-799. Fix CgroupsLCEResourcesHandler to use /tasks instead of
     /cgroup.procs. (Chris Riccomini via acmurthy) 
 
+    YARN-333. Schedulers cannot control the queue-name of an 
+    application. (sandyr via tucu)
+
 Release 2.0.5-alpha - 06/06/2013
 
   INCOMPATIBLE CHANGES

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

@@ -33,7 +33,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttempt;
 import org.apache.hadoop.yarn.server.resourcemanager.rmnode.RMNode;
 
 /**
- * The read interface to an Application in the ResourceManager. Take a
+ * The interface to an Application in the ResourceManager. Take a
  * look at {@link RMAppImpl} for its implementation. This interface
  * exposes methods to access various updates in application status/report.
  */
@@ -85,6 +85,13 @@ public interface RMApp extends EventHandler<RMAppEvent> {
    * @return the queue to which the application was submitted to.
    */
   String getQueue();
+  
+  /**
+   * Reflects a change in the application's queue from the one specified in the
+   * {@link ApplicationSubmissionContext}.
+   * @param name the new queue name
+   */
+  void setQueue(String name);
 
   /**
    * The name of the application as set in {@link

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

@@ -81,7 +81,6 @@ public class RMAppImpl implements RMApp, Recoverable {
   private final RMContext rmContext;
   private final Configuration conf;
   private final String user;
-  private final String queue;
   private final String name;
   private final ApplicationSubmissionContext submissionContext;
   private final Dispatcher dispatcher;
@@ -101,6 +100,7 @@ public class RMAppImpl implements RMApp, Recoverable {
   private long startTime;
   private long finishTime;
   private RMAppAttempt currentAttempt;
+  private String queue;
   @SuppressWarnings("rawtypes")
   private EventHandler handler;
   private static final FinalTransition FINAL_TRANSITION = new FinalTransition();
@@ -341,6 +341,11 @@ public class RMAppImpl implements RMApp, Recoverable {
   public String getQueue() {
     return this.queue;
   }
+  
+  @Override
+  public void setQueue(String queue) {
+    this.queue = queue;
+  }
 
   @Override
   public String getName() {

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

@@ -57,6 +57,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.RMAuditLogger.AuditConstant
 import org.apache.hadoop.yarn.server.resourcemanager.RMContext;
 import org.apache.hadoop.yarn.server.resourcemanager.recovery.RMStateStore.RMState;
 import org.apache.hadoop.yarn.server.resourcemanager.resource.ResourceWeights;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMApp;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEvent;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptEventType;
 import org.apache.hadoop.yarn.server.resourcemanager.rmapp.attempt.RMAppAttemptState;
@@ -88,6 +89,8 @@ import org.apache.hadoop.yarn.util.resource.DominantResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.ResourceCalculator;
 import org.apache.hadoop.yarn.util.resource.Resources;
 
+import com.google.common.annotations.VisibleForTesting;
+
 /**
  * A scheduler that schedules resources between a set of queues. The scheduler
  * keeps track of the resources used by each queue, and attempts to maintain
@@ -601,12 +604,8 @@ public class FairScheduler implements ResourceScheduler {
    */
   protected synchronized void addApplication(
       ApplicationAttemptId applicationAttemptId, String queueName, String user) {
-
-    FSLeafQueue queue = queueMgr.getLeafQueue(queueName);
-    if (queue == null) {
-      // queue is not an existing or createable leaf queue
-      queue = queueMgr.getLeafQueue(YarnConfiguration.DEFAULT_QUEUE_NAME);
-    }
+    RMApp rmApp = rmContext.getRMApps().get(applicationAttemptId);
+    FSLeafQueue queue = assignToQueue(rmApp, queueName, user);
 
     FSSchedulerApp schedulerApp =
         new FSSchedulerApp(applicationAttemptId, user,
@@ -637,6 +636,27 @@ public class FairScheduler implements ResourceScheduler {
         new RMAppAttemptEvent(applicationAttemptId,
             RMAppAttemptEventType.APP_ACCEPTED));
   }
+  
+  @VisibleForTesting
+  FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) {
+    // Potentially set queue to username if configured to do so
+    if (queueName.equals(YarnConfiguration.DEFAULT_QUEUE_NAME) &&
+        userAsDefaultQueue) {
+      queueName = user;
+    }
+    
+    FSLeafQueue queue = queueMgr.getLeafQueue(queueName);
+    if (queue == null) {
+      // queue is not an existing or createable leaf queue
+      queue = queueMgr.getLeafQueue(YarnConfiguration.DEFAULT_QUEUE_NAME);
+    }
+    
+    if (rmApp != null) {
+      rmApp.setQueue(queue.getName());
+    }
+    
+    return queue;
+  }
 
   private synchronized void removeApplication(
       ApplicationAttemptId applicationAttemptId,
@@ -985,13 +1005,6 @@ public class FairScheduler implements ResourceScheduler {
       }
       AppAddedSchedulerEvent appAddedEvent = (AppAddedSchedulerEvent)event;
       String queue = appAddedEvent.getQueue();
-
-      // Potentially set queue to username if configured to do so
-      String def = YarnConfiguration.DEFAULT_QUEUE_NAME;
-      if (queue.equals(def) && userAsDefaultQueue) {
-        queue = appAddedEvent.getUser();
-      }
-
       addApplication(appAddedEvent.getApplicationAttemptId(), queue,
           appAddedEvent.getUser());
       break;

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

@@ -136,6 +136,11 @@ public abstract class MockAsm extends MockApps {
     public String getApplicationType() {
       throw new UnsupportedOperationException("Not supported yet.");
     }
+
+    @Override
+    public void setQueue(String name) {
+      throw new UnsupportedOperationException("Not supported yet.");
+    }
   }
 
   public static RMApp newApplication(int i) {

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

@@ -63,6 +63,7 @@ import org.apache.hadoop.yarn.server.resourcemanager.ApplicationMasterService;
 import org.apache.hadoop.yarn.server.resourcemanager.MockNodes;
 import org.apache.hadoop.yarn.server.resourcemanager.RMContextImpl;
 import org.apache.hadoop.yarn.server.resourcemanager.ResourceManager;
+import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp;
 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.RMAppEventType;
@@ -393,7 +394,6 @@ public class TestFairScheduler {
     scheduler.handle(nodeEvent1);
 
     // Have two queues which want entire cluster capacity
-    createSchedulingRequest(10 * 1024, "default", "user1");
     createSchedulingRequest(10 * 1024, "parent.queue2", "user1");
     createSchedulingRequest(10 * 1024, "parent.queue3", "user1");
 
@@ -558,6 +558,25 @@ public class TestFairScheduler {
     assertEquals(0, scheduler.getQueueManager().getLeafQueue("user2")
         .getAppSchedulables().size());
   }
+  
+  @Test
+  public void testAssignToQueue() throws Exception {
+    Configuration conf = createConfiguration();
+    conf.set(FairSchedulerConfiguration.USER_AS_DEFAULT_QUEUE, "true");
+    scheduler.reinitialize(conf, resourceManager.getRMContext());
+    
+    RMApp rmApp1 = new MockRMApp(0, 0, RMAppState.NEW);
+    RMApp rmApp2 = new MockRMApp(1, 1, RMAppState.NEW);
+    
+    FSLeafQueue queue1 = scheduler.assignToQueue(rmApp1, "default", "asterix");
+    FSLeafQueue queue2 = scheduler.assignToQueue(rmApp2, "notdefault", "obelix");
+    
+    // assert FSLeafQueue's name is the correct name is the one set in the RMApp
+    assertEquals(rmApp1.getQueue(), queue1.getName());
+    assertEquals("root.asterix", rmApp1.getQueue());
+    assertEquals(rmApp2.getQueue(), queue2.getName());
+    assertEquals("root.notdefault", rmApp2.getQueue());
+  }
 
   @Test
   public void testFairShareWithMinAlloc() throws Exception {