CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable { @Override public List<ApplicationAttemptId> getAppsInQueue(String queueName) { CSQueue queue = queues.get(queueName); if (queue == null) { return null; } List<ApplicationAttemptId> apps = new ArrayList<ApplicationAttemptId>(); queue.collectSchedulerApplications(apps); return apps; } CapacityScheduler(); @Override void setConf(Configuration conf); @Override boolean isDockerMonitorEnabled(); @Override Configuration getConf(); @VisibleForTesting synchronized String getMappedQueueForTest(String user); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override Comparator<FiCaSchedulerApp> getApplicationComparator(); @Override ResourceCalculator getResourceCalculator(); @Override Comparator<CSQueue> getQueueComparator(); @Override int getNumClusterNodes(); @Override synchronized RMContext getRMContext(); @Override synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @VisibleForTesting static void setQueueAcls(YarnAuthorizationProvider authorizer,
      Map<String, CSQueue> queues); CSQueue getQueue(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release, 
      List<String> blacklistAdditions, List<String> blacklistRemovals); @Override @Lock(Lock.NoLock.class) QueueInfo getQueueInfo(String queueName, 
      boolean includeChildQueues, boolean recursive); @Override @Lock(Lock.NoLock.class) List<QueueUserACLInfo> getQueueUserAclInfo(); @Override void handle(SchedulerEvent event); @Lock(Lock.NoLock.class) @VisibleForTesting @Override FiCaSchedulerApp getApplicationAttempt(
      ApplicationAttemptId applicationAttemptId); @Lock(Lock.NoLock.class) FiCaSchedulerNode getNode(NodeId nodeId); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void dropContainerReservation(RMContainer container); @Override void preemptContainer(ApplicationAttemptId aid, RMContainer cont); @Override void killContainer(RMContainer cont); void resumeContainer(RMContainer cont); @Override void suspendContianer(RMContainer cont, Resource toPreempt); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized void removeQueue(String queueName); @Override synchronized void addQueue(Queue queue); @Override synchronized void setEntitlement(String inQueue,
      QueueEntitlement entitlement); @Override synchronized String moveApplication(ApplicationId appId,
      String targetQueueName); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); @Override List<NodeContainerUpdate> pullNodeContainerUpdate(NodeId node); @Private
  static final String ROOT_QUEUE; }
@Test public void testGetAppsInQueue() throws Exception { Application application_0 = new Application("user_0", "a1", resourceManager); application_0.submit(); Application application_1 = new Application("user_0", "a2", resourceManager); application_1.submit(); Application application_2 = new Application("user_0", "b2", resourceManager); application_2.submit(); ResourceScheduler scheduler = resourceManager.getResourceScheduler(); List<ApplicationAttemptId> appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); List<ApplicationAttemptId> appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(application_0.getApplicationAttemptId())); assertTrue(appsInA.contains(application_1.getApplicationAttemptId())); assertEquals(2, appsInA.size()); List<ApplicationAttemptId> appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(application_0.getApplicationAttemptId())); assertTrue(appsInRoot.contains(application_1.getApplicationAttemptId())); assertTrue(appsInRoot.contains(application_2.getApplicationAttemptId())); assertEquals(3, appsInRoot.size()); Assert.assertNull(scheduler.getAppsInQueue("nonexistentqueue")); }
@Test public void testMoveAllAppsInvalidDestination() throws Exception { MockRM rm = setUpMove(); AbstractYarnScheduler scheduler = (AbstractYarnScheduler) rm.getResourceScheduler(); RMApp app = rm.submitApp(GB, "test-move-1", "user_0", null, "a1"); ApplicationAttemptId appAttemptId = rm.getApplicationReport(app.getApplicationId()) .getCurrentApplicationAttemptId(); List<ApplicationAttemptId> appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); List<ApplicationAttemptId> appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); List<ApplicationAttemptId> appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); List<ApplicationAttemptId> appsInB1 = scheduler.getAppsInQueue("b1"); assertTrue(appsInB1.isEmpty()); List<ApplicationAttemptId> appsInB = scheduler.getAppsInQueue("b"); assertTrue(appsInB.isEmpty()); try { scheduler.moveAllApps("a1", "DOES_NOT_EXIST"); Assert.fail(); } catch (YarnException e) { } appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); appsInB1 = scheduler.getAppsInQueue("b1"); assertTrue(appsInB1.isEmpty()); appsInB = scheduler.getAppsInQueue("b"); assertTrue(appsInB.isEmpty()); rm.stop(); }
@Test public void testMoveAllAppsInvalidSource() throws Exception { MockRM rm = setUpMove(); AbstractYarnScheduler scheduler = (AbstractYarnScheduler) rm.getResourceScheduler(); RMApp app = rm.submitApp(GB, "test-move-1", "user_0", null, "a1"); ApplicationAttemptId appAttemptId = rm.getApplicationReport(app.getApplicationId()) .getCurrentApplicationAttemptId(); List<ApplicationAttemptId> appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); List<ApplicationAttemptId> appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); List<ApplicationAttemptId> appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); List<ApplicationAttemptId> appsInB1 = scheduler.getAppsInQueue("b1"); assertTrue(appsInB1.isEmpty()); List<ApplicationAttemptId> appsInB = scheduler.getAppsInQueue("b"); assertTrue(appsInB.isEmpty()); try { scheduler.moveAllApps("DOES_NOT_EXIST", "b1"); Assert.fail(); } catch (YarnException e) { } appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); appsInB1 = scheduler.getAppsInQueue("b1"); assertTrue(appsInB1.isEmpty()); appsInB = scheduler.getAppsInQueue("b"); assertTrue(appsInB.isEmpty()); rm.stop(); }
@Test public void testKillAllAppsInvalidSource() throws Exception { MockRM rm = setUpMove(); AbstractYarnScheduler scheduler = (AbstractYarnScheduler) rm.getResourceScheduler(); RMApp app = rm.submitApp(GB, "test-move-1", "user_0", null, "a1"); ApplicationAttemptId appAttemptId = rm.getApplicationReport(app.getApplicationId()) .getCurrentApplicationAttemptId(); List<ApplicationAttemptId> appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); List<ApplicationAttemptId> appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); List<ApplicationAttemptId> appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); try { scheduler.killAllAppsInQueue("DOES_NOT_EXIST"); Assert.fail(); } catch (YarnException e) { } appsInA1 = scheduler.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); appsInA = scheduler.getAppsInQueue("a"); assertTrue(appsInA.contains(appAttemptId)); assertEquals(1, appsInA.size()); appsInRoot = scheduler.getAppsInQueue("root"); assertTrue(appsInRoot.contains(appAttemptId)); assertEquals(1, appsInRoot.size()); rm.stop(); }