CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler { @Override @Lock(Lock.NoLock.class) public Allocation allocate(ApplicationAttemptId applicationAttemptId, List<ResourceRequest> ask, List<SchedulingRequest> schedulingRequests, List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals, ContainerUpdates updateRequests) { FiCaSchedulerApp application = getApplicationAttempt(applicationAttemptId); if (application == null) { LOG.error("Calling allocate on removed or non existent application " + applicationAttemptId.getApplicationId()); return EMPTY_ALLOCATION; } if (!application.getApplicationAttemptId().equals(applicationAttemptId)) { LOG.error("Calling allocate on previous or removed " + "or non existent application attempt " + applicationAttemptId); return EMPTY_ALLOCATION; } handleContainerUpdates(application, updateRequests); releaseContainers(release, application); LeafQueue updateDemandForQueue = null; normalizeResourceRequests(ask); normalizeSchedulingRequests(schedulingRequests); Allocation allocation; try { application.getWriteLock().lock(); if (application.isStopped()) { return EMPTY_ALLOCATION; } if (!ask.isEmpty() || (schedulingRequests != null && !schedulingRequests .isEmpty())) { if (LOG.isDebugEnabled()) { LOG.debug( "allocate: pre-update " + applicationAttemptId + " ask size =" + ask.size()); application.showRequests(); } if (application.updateResourceRequests(ask) || application .updateSchedulingRequests(schedulingRequests)) { updateDemandForQueue = (LeafQueue) application.getQueue(); } if (LOG.isDebugEnabled()) { LOG.debug("allocate: post-update"); application.showRequests(); } } application.updateBlacklist(blacklistAdditions, blacklistRemovals); allocation = application.getAllocation(getResourceCalculator(), getClusterResource(), getMinimumResourceCapability()); } finally { application.getWriteLock().unlock(); } if (updateDemandForQueue != null && !application .isWaitingForAMContainer()) { updateDemandForQueue.getOrderingPolicy().demandUpdated(application); } if (LOG.isDebugEnabled()) { LOG.info("Allocation for application " + applicationAttemptId + " : " + allocation + " with cluster resource : " + getClusterResource()); } return allocation; } 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 testPendingResourceUpdatedAccordingToIncreaseRequestChanges() throws Exception { Configuration conf = TestUtils.getConfigurationWithQueueLabels(new Configuration(false)); conf.setBoolean(YarnConfiguration.NODE_LABELS_ENABLED, true); final RMNodeLabelsManager mgr = new NullRMNodeLabelsManager(); mgr.init(conf); MockRM rm = new MockRM(conf) { protected RMNodeLabelsManager createNodeLabelManager() { return mgr; } }; rm.start(); MockNM nm1 = new MockNM("h1:1234", 200 * GB, rm.getResourceTrackerService()); nm1.registerNode(); RMApp app1 = rm.submitApp(1 * GB, "app", "user", null, "a1"); MockAM am1 = MockRM.launchAndRegisterAM(app1, rm, nm1); am1.allocate( Arrays.asList(ResourceRequest.newInstance(Priority.newInstance(1), "*", Resources.createResource(2 * GB), 2)), null); ContainerId containerId1 = ContainerId.newContainerId(am1.getApplicationAttemptId(), 1); ContainerId containerId2 = ContainerId.newContainerId(am1.getApplicationAttemptId(), 2); ContainerId containerId3 = ContainerId.newContainerId(am1.getApplicationAttemptId(), 3); Assert.assertTrue(rm.waitForState(nm1, containerId3, RMContainerState.ALLOCATED)); am1.allocate(null, null); sentRMContainerLaunched(rm, ContainerId.newContainerId(am1.getApplicationAttemptId(), 1L)); sentRMContainerLaunched(rm, ContainerId.newContainerId(am1.getApplicationAttemptId(), 2L)); sentRMContainerLaunched(rm, ContainerId.newContainerId(am1.getApplicationAttemptId(), 3L)); am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest .newInstance(0, containerId1, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(3 * GB), null))); FiCaSchedulerApp app = getFiCaSchedulerApp(rm, app1.getApplicationId()); Assert.assertEquals(2 * GB, app.getAppAttemptResourceUsage().getPending().getMemorySize()); checkPendingResource(rm, "a1", 2 * GB, null); checkPendingResource(rm, "a", 2 * GB, null); checkPendingResource(rm, "root", 2 * GB, null); am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest .newInstance(0, containerId2, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(3 * GB), null), UpdateContainerRequest .newInstance(0, containerId3, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(5 * GB), null))); Assert.assertEquals(6 * GB, app.getAppAttemptResourceUsage().getPending().getMemorySize()); checkPendingResource(rm, "a1", 6 * GB, null); checkPendingResource(rm, "a", 6 * GB, null); checkPendingResource(rm, "root", 6 * GB, null); am1.sendContainerResizingRequest(Arrays.asList( UpdateContainerRequest .newInstance(0, containerId1, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(3 * GB), null), UpdateContainerRequest .newInstance(0, containerId2, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(4 * GB), null), UpdateContainerRequest .newInstance(0, containerId3, ContainerUpdateType.INCREASE_RESOURCE, Resources.createResource(2 * GB), null))); Assert.assertEquals(4 * GB, app.getAppAttemptResourceUsage().getPending().getMemorySize()); checkPendingResource(rm, "a1", 4 * GB, null); checkPendingResource(rm, "a", 4 * GB, null); checkPendingResource(rm, "root", 4 * GB, null); }