@Test public void testAssignToQueue() throws Exception { conf.set(FairSchedulerConfiguration.USER_AS_DEFAULT_QUEUE, "true"); scheduler.init(conf); scheduler.start(); 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"); assertEquals(rmApp1.getQueue(), queue1.getName()); assertEquals("root.asterix", rmApp1.getQueue()); assertEquals(rmApp2.getQueue(), queue2.getName()); assertEquals("root.notdefault", rmApp2.getQueue()); }
@VisibleForTesting FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) { FSLeafQueue queue = null; String appRejectMsg = null; try { QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy(); queueName = placementPolicy.assignAppToQueue(queueName, user); if (queueName == null) { appRejectMsg = "Application rejected by queue placement policy"; } else { queue = queueMgr.getLeafQueue(queueName, true); if (queue == null) { appRejectMsg = queueName + " is not a leaf queue"; } } } catch (IOException ioe) { appRejectMsg = "Error assigning app to queue " + queueName; } if (appRejectMsg != null && rmApp != null) { LOG.error(appRejectMsg); rmContext.getDispatcher().getEventHandler().handle( new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg)); return null; } if (rmApp != null) { rmApp.setQueue(queue.getName()); } else { LOG.error("Couldn't find RM app to set queue name on"); } return queue; }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { @VisibleForTesting FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) { FSLeafQueue queue = null; String appRejectMsg = null; try { QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy(); queueName = placementPolicy.assignAppToQueue(queueName, user); if (queueName == null) { appRejectMsg = "Application rejected by queue placement policy"; } else { queue = queueMgr.getLeafQueue(queueName, true); if (queue == null) { appRejectMsg = queueName + " is not a leaf queue"; } } } catch (IOException ioe) { appRejectMsg = "Error assigning app to queue " + queueName; } if (appRejectMsg != null && rmApp != null) { LOG.error(appRejectMsg); rmContext.getDispatcher().getEventHandler().handle( new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg)); return null; } if (rmApp != null) { rmApp.setQueue(queue.getName()); } else { LOG.error("Couldn't find RM app to set queue name on"); } return queue; } }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { @VisibleForTesting FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) { FSLeafQueue queue = null; String appRejectMsg = null; try { QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy(); queueName = placementPolicy.assignAppToQueue(queueName, user); if (queueName == null) { appRejectMsg = "Application rejected by queue placement policy"; } else { queue = queueMgr.getLeafQueue(queueName, true); if (queue == null) { appRejectMsg = queueName + " is not a leaf queue"; } } } catch (IOException ioe) { appRejectMsg = "Error assigning app to queue " + queueName; } if (appRejectMsg != null && rmApp != null) { LOG.error(appRejectMsg); rmContext.getDispatcher().getEventHandler().handle( new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg)); return null; } if (rmApp != null) { rmApp.setQueue(queue.getName()); } else { LOG.error("Couldn't find RM app to set queue name on"); } return queue; } FairScheduler(); }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { @VisibleForTesting FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) { FSLeafQueue queue = null; String appRejectMsg = null; try { QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy(); queueName = placementPolicy.assignAppToQueue(queueName, user); if (queueName == null) { appRejectMsg = "Application rejected by queue placement policy"; } else { queue = queueMgr.getLeafQueue(queueName, true); if (queue == null) { appRejectMsg = queueName + " is not a leaf queue"; } } } catch (IOException ioe) { appRejectMsg = "Error assigning app to queue " + queueName; } if (appRejectMsg != null && rmApp != null) { LOG.error(appRejectMsg); rmContext.getDispatcher().getEventHandler().handle( new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg)); return null; } if (rmApp != null) { rmApp.setQueue(queue.getName()); } else { LOG.error("Couldn't find RM app to set queue name on"); } return queue; } FairScheduler(); FairSchedulerConfiguration getConf(); QueueManager getQueueManager(); synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); synchronized ResourceWeights getAppWeight(FSAppAttempt app); Resource getIncrementResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); long getNodeLocalityDelayMs(); long getRackLocalityDelayMs(); boolean isContinuousSchedulingEnabled(); synchronized int getContinuousSchedulingSleepMs(); Clock getClock(); FairSchedulerEventLog getEventLog(); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release,
      List<String> blacklistAdditions, List<String> blacklistRemovals); FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId); @Override ResourceCalculator getResourceCalculator(); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); AllocationConfiguration getAllocationConfiguration(); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized String moveApplication(ApplicationId appId,
      String queueName); @Override synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); @Override void setEntitlement(String queueName,
      QueueEntitlement entitlement); @Override void removeQueue(String queueName); }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { @VisibleForTesting FSLeafQueue assignToQueue(RMApp rmApp, String queueName, String user) { FSLeafQueue queue = null; String appRejectMsg = null; try { QueuePlacementPolicy placementPolicy = allocConf.getPlacementPolicy(); queueName = placementPolicy.assignAppToQueue(queueName, user); if (queueName == null) { appRejectMsg = "Application rejected by queue placement policy"; } else { queue = queueMgr.getLeafQueue(queueName, true); if (queue == null) { appRejectMsg = queueName + " is not a leaf queue"; } } } catch (IOException ioe) { appRejectMsg = "Error assigning app to queue " + queueName; } if (appRejectMsg != null && rmApp != null) { LOG.error(appRejectMsg); rmContext.getDispatcher().getEventHandler().handle( new RMAppRejectedEvent(rmApp.getApplicationId(), appRejectMsg)); return null; } if (rmApp != null) { rmApp.setQueue(queue.getName()); } else { LOG.error("Couldn't find RM app to set queue name on"); } return queue; } FairScheduler(); FairSchedulerConfiguration getConf(); QueueManager getQueueManager(); synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); synchronized ResourceWeights getAppWeight(FSAppAttempt app); Resource getIncrementResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); long getNodeLocalityDelayMs(); long getRackLocalityDelayMs(); boolean isContinuousSchedulingEnabled(); synchronized int getContinuousSchedulingSleepMs(); Clock getClock(); FairSchedulerEventLog getEventLog(); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release,
      List<String> blacklistAdditions, List<String> blacklistRemovals); FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId); @Override ResourceCalculator getResourceCalculator(); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); AllocationConfiguration getAllocationConfiguration(); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized String moveApplication(ApplicationId appId,
      String queueName); @Override synchronized void updateNodeResource(RMNode nm, 
      ResourceOption resourceOption); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Set<String> getPlanQueues(); @Override void setEntitlement(String queueName,
      QueueEntitlement entitlement); @Override void removeQueue(String queueName); static final Resource CONTAINER_RESERVED; }