CapacityScheduler extends
    AbstractYarnScheduler<FiCaSchedulerApp, FiCaSchedulerNode> implements
    PreemptableResourceScheduler, CapacitySchedulerContext, Configurable,
    ResourceAllocationCommitter, MutableConfScheduler { public CapacityScheduler() { super(CapacityScheduler.class.getName()); this.maxRunningEnforcer = new CSMaxRunningAppsEnforcer(this); } 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(); void reinitialize(Configuration newConf, RMContext rmContext,
         boolean validation); @Override void reinitialize(Configuration newConf, RMContext rmContext); @VisibleForTesting PlacementRule getUserGroupMappingPlacementRule(); PlacementRule getAppNameMappingPlacementRule(); @VisibleForTesting void updatePlacementRules(); @Override CSQueue getQueue(String queueName); String normalizeQueueName(String name); boolean isAmbiguous(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 queuePath, 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(); WorkflowPriorityMappingsManager getWorkflowPriorityMappingsManager(); 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(); boolean isMultiNodePlacementEnabled(); int getNumAsyncSchedulerThreads(); @VisibleForTesting void setMaxRunningAppsEnforcer(CSMaxRunningAppsEnforcer enforcer); @Override boolean placementConstraintEnabled(); @VisibleForTesting void setActivitiesManagerEnabled(boolean enabled);  }
@Test public void testCapacityScheduler() throws Exception { LOG.info("--- START: testCapacityScheduler ---"); NodeStatus mockNodeStatus = createMockNodeStatus(); String host_0 = "host_0"; NodeManager nm_0 = registerNode(host_0, 1234, 2345, NetworkTopology.DEFAULT_RACK, Resources.createResource(4 * GB, 1), mockNodeStatus); String host_1 = "host_1"; NodeManager nm_1 = registerNode(host_1, 1234, 2345, NetworkTopology.DEFAULT_RACK, Resources.createResource(2 * GB, 1), mockNodeStatus); Priority priority_0 = Priority.newInstance(0); Priority priority_1 = Priority.newInstance(1); Application application_0 = new Application("user_0", "a1", resourceManager); application_0.submit(); application_0.addNodeManager(host_0, 1234, nm_0); application_0.addNodeManager(host_1, 1234, nm_1); Resource capability_0_0 = Resources.createResource(1 * GB, 1); application_0.addResourceRequestSpec(priority_1, capability_0_0); Resource capability_0_1 = Resources.createResource(2 * GB, 1); application_0.addResourceRequestSpec(priority_0, capability_0_1); Task task_0_0 = new Task(application_0, priority_1, new String[] {host_0, host_1}); application_0.addTask(task_0_0); Application application_1 = new Application("user_1", "b2", resourceManager); application_1.submit(); application_1.addNodeManager(host_0, 1234, nm_0); application_1.addNodeManager(host_1, 1234, nm_1); Resource capability_1_0 = Resources.createResource(3 * GB, 1); application_1.addResourceRequestSpec(priority_1, capability_1_0); Resource capability_1_1 = Resources.createResource(2 * GB, 1); application_1.addResourceRequestSpec(priority_0, capability_1_1); Task task_1_0 = new Task(application_1, priority_1, new String[] {host_0, host_1}); application_1.addTask(task_1_0); application_0.schedule(); application_1.schedule(); LOG.info("Kick!"); nodeUpdate(nm_0); nodeUpdate(nm_1); application_0.schedule(); checkApplicationResourceUsage(1 * GB, application_0); application_1.schedule(); checkApplicationResourceUsage(3 * GB, application_1); checkNodeResourceUsage(4*GB, nm_0); checkNodeResourceUsage(0*GB, nm_1); LOG.info("Adding new tasks..."); Task task_1_1 = new Task(application_1, priority_0, new String[] {ResourceRequest.ANY}); application_1.addTask(task_1_1); application_1.schedule(); Task task_0_1 = new Task(application_0, priority_0, new String[] {host_0, host_1}); application_0.addTask(task_0_1); application_0.schedule(); LOG.info("Sending hb from " + nm_0.getHostName()); nodeUpdate(nm_0); LOG.info("Sending hb from " + nm_1.getHostName()); nodeUpdate(nm_1); LOG.info("Trying to allocate..."); application_0.schedule(); checkApplicationResourceUsage(1 * GB, application_0); application_1.schedule(); checkApplicationResourceUsage(5 * GB, application_1); nodeUpdate(nm_0); nodeUpdate(nm_1); checkNodeResourceUsage(4*GB, nm_0); checkNodeResourceUsage(2*GB, nm_1); LOG.info("--- END: testCapacityScheduler ---"); }