CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler { @Override public List<ApplicationAttemptId> getAppsInQueue(String queueName) { CSQueue queue = getQueue(queueName); if (queue == null) { return null; } List<ApplicationAttemptId> apps = new ArrayList<ApplicationAttemptId>(); queue.collectSchedulerApplications(apps); return apps; } CapacityScheduler(); @Override void setConf(Configuration conf); @Override Configuration getConf(); @Override QueueMetrics getRootQueueMetrics(); CSQueue getRootQueue(); @Override CapacitySchedulerConfiguration getConfiguration(); @Override RMContainerTokenSecretManager getContainerTokenSecretManager(); @Override ResourceCalculator getResourceCalculator(); @VisibleForTesting void setResourceCalculator(ResourceCalculator rc); @Override int getNumClusterNodes(); @Override RMContext getRMContext(); @Override void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override void reinitialize(Configuration newConf, RMContext rmContext); @VisibleForTesting PlacementRule getUserGroupMappingPlacementRule(); PlacementRule getAppNameMappingPlacementRule(); @VisibleForTesting void updatePlacementRules(); @Override CSQueue getQueue(String queueName); @Override @Lock(Lock.NoLock.class) Allocation allocate(ApplicationAttemptId applicationAttemptId,
      List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests,
      List<ContainerId> release, List<String> blacklistAdditions,
      List<String> blacklistRemovals, ContainerUpdates updateRequests); @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); @Lock(Lock.NoLock.class) List<FiCaSchedulerNode> getAllNodes(); @Override @Lock(Lock.NoLock.class) void recover(RMState state); @Override void killReservedContainer(RMContainer container); @Override void markContainerForPreemption(ApplicationAttemptId aid,
      RMContainer cont); @VisibleForTesting @Override void killContainer(RMContainer container); void markContainerForKillable(
      RMContainer killableContainer); @Override boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); boolean isSystemAppsLimitReached(); @Override void removeQueue(String queueName); @Override void addQueue(Queue queue); @Override void setEntitlement(String inQueue, QueueEntitlement entitlement); @Override String moveApplication(ApplicationId appId,
      String targetQueueName); @Override void preValidateMoveApplication(ApplicationId appId,
      String newQueue); @Override EnumSet<SchedulerResourceTypes> getSchedulingResourceTypes(); @Override Resource getMaximumResourceCapability(String queueName); @Override Set<String> getPlanQueues(); @Override Priority checkAndGetApplicationPriority(
      Priority priorityRequestedByApp, UserGroupInformation user,
      String queueName, ApplicationId applicationId); @Override Priority updateApplicationPriority(Priority newPriority,
      ApplicationId applicationId, SettableFuture<Object> future,
      UserGroupInformation user); @Override PreemptionManager getPreemptionManager(); @Override ResourceUsage getClusterResourceUsage(); @VisibleForTesting void submitResourceCommitRequest(Resource cluster,
      CSAssignment csAssignment); @Override boolean attemptAllocationOnNode(SchedulerApplicationAttempt appAttempt,
      SchedulingRequest schedulingRequest, SchedulerNode schedulerNode); @VisibleForTesting ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> createResourceCommitRequest(CSAssignment csAssignment); @Override boolean tryCommit(Resource cluster, ResourceCommitRequest r,
      boolean updatePending); int getAsyncSchedulingPendingBacklogs(); @Override CapacitySchedulerQueueManager getCapacitySchedulerQueueManager(); boolean moveReservedContainer(RMContainer toBeMovedContainer,
      FiCaSchedulerNode targetNode); @Override long checkAndGetApplicationLifetime(String queueName,
      long lifetimeRequestedByApp); @Override long getMaximumApplicationLifetime(String queueName); @Override boolean isConfigurationMutable(); @Override MutableConfigurationProvider getMutableConfProvider(); @Override void resetSchedulerMetrics();  }
@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(); YarnScheduler scheduler = 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(); YarnScheduler scheduler = 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(); YarnScheduler scheduler = 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(); }
@Test public void testQueueMappingWithCurrentUserQueueMappingForaGroup() throws Exception { CapacitySchedulerConfiguration config = new CapacitySchedulerConfiguration(); config.setClass(YarnConfiguration.RM_SCHEDULER, CapacityScheduler.class, ResourceScheduler.class); setupQueueConfiguration(config); config.setClass(CommonConfigurationKeys.HADOOP_SECURITY_GROUP_MAPPING, TestGroupsCaching.FakeunPrivilegedGroupMapping.class, ShellBasedUnixGroupsMapping.class); config.set(CommonConfigurationKeys.HADOOP_USER_GROUP_STATIC_OVERRIDES, "a1" +"=" + "agroup" + ""); config.set(CapacitySchedulerConfiguration.QUEUE_MAPPING, "g:agroup:%user"); MockRM rm = new MockRM(config); rm.start(); CapacityScheduler cs = ((CapacityScheduler) rm.getResourceScheduler()); cs.start(); RMApp app = rm.submitApp(GB, "appname", "a1", null, "default"); List<ApplicationAttemptId> appsInA1 = cs.getAppsInQueue("a1"); assertEquals(1, appsInA1.size()); }