LeafQueue extends AbstractCSQueue { @Lock({LeafQueue.class}) Resource computeUserLimitAndSetHeadroom(FiCaSchedulerApp application, Resource clusterResource, String nodePartition, SchedulingMode schedulingMode, Resource userLimit) { String user = application.getUser(); User queueUser = getUser(user); if (queueUser == null) { LOG.debug("User {} has been removed!", user); return Resources.none(); } if (userLimit == null) { userLimit = getResourceLimitForActiveUsers(application.getUser(), clusterResource, nodePartition, schedulingMode); } setQueueResourceLimitsInfo(clusterResource); Resource headroom = metrics.getUserMetrics(user) == null ? Resources.none() : getHeadroom(queueUser, cachedResourceLimitsForHeadroom.getLimit(), clusterResource, userLimit, nodePartition); if (LOG.isDebugEnabled()) { LOG.debug("Headroom calculation for user " + user + ": " + " userLimit=" + userLimit + " queueMaxAvailRes=" + cachedResourceLimitsForHeadroom.getLimit() + " consumed=" + queueUser.getUsed() + " partition=" + nodePartition); } CapacityHeadroomProvider headroomProvider = new CapacityHeadroomProvider( queueUser, this, application, queueResourceLimitsInfo); application.setHeadroomProvider(headroomProvider); metrics.setAvailableResourcesToUser(nodePartition, user, headroom); return userLimit; } @SuppressWarnings({ "unchecked", "rawtypes" })  LeafQueue(CapacitySchedulerContext cs,
      String queueName, CSQueue parent, CSQueue old);  LeafQueue(CapacitySchedulerContext cs,
      CapacitySchedulerConfiguration configuration,
      String queueName, CSQueue parent, CSQueue old); @Private float getMinimumAllocationFactor(); @Private float getMaxAMResourcePerQueuePercent(); int getMaxApplications(); int getMaxApplicationsPerUser(); UsersManager getUsersManager(); @Override AbstractUsersManager getAbstractUsersManager(); @Override List<CSQueue> getChildQueues(); @Override int getNumApplications(); int getNumPendingApplications(); int getNumActiveApplications(); @Private int getNumPendingApplications(String user); @Private int getNumActiveApplications(String user); @Private int getUserLimit(); @Private float getUserLimitFactor(); @Override QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(UserGroupInformation user); String toString(); @VisibleForTesting User getUser(String userName); @Private List<AppPriorityACLGroup> getPriorityACLs(); @Override void reinitialize(
      CSQueue newlyParsedQueue, Resource clusterResource); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName, boolean isMoveApp); @Override void submitApplication(ApplicationId applicationId, String userName,
      String queue); void validateSubmitApplication(ApplicationId applicationId,
      String userName, String queue); Resource getAMResourceLimit(); Resource getAMResourceLimitPerPartition(String nodePartition); @VisibleForTesting Resource calculateAndGetAMResourceLimit(); @VisibleForTesting Resource getUserAMResourceLimit(); Resource getUserAMResourceLimitPerPartition(
      String nodePartition, String userName); Resource calculateAndGetAMResourceLimitPerPartition(
      String nodePartition); @Override void finishApplication(ApplicationId application, String user); @Override void finishApplicationAttempt(FiCaSchedulerApp application, String queue); @Override CSAssignment assignContainers(Resource clusterResource,
      CandidateNodeSet<FiCaSchedulerNode> candidates,
      ResourceLimits currentResourceLimits, SchedulingMode schedulingMode); @Override boolean accept(Resource cluster,
      ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> request); void apply(Resource cluster,
      ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> request); @Lock(NoLock.class) int getNodeLocalityDelay(); @Lock(NoLock.class) int getRackLocalityAdditionalDelay(); @Lock(NoLock.class) boolean getRackLocalityFullReset(); Resource getResourceLimitForActiveUsers(String userName,
      Resource clusterResource, String nodePartition,
      SchedulingMode schedulingMode); Resource getResourceLimitForAllUsers(String userName,
      Resource clusterResource, String nodePartition,
      SchedulingMode schedulingMode); void recalculateQueueUsageRatio(Resource clusterResource,
      String nodePartition); @Override void completedContainer(Resource clusterResource, 
      FiCaSchedulerApp application, FiCaSchedulerNode node, RMContainer rmContainer, 
      ContainerStatus containerStatus, RMContainerEventType event, CSQueue childQueue,
      boolean sortQueues); @Override void updateClusterResource(Resource clusterResource,
      ResourceLimits currentResourceLimits); @Override void incUsedResource(String nodeLabel, Resource resourceToInc,
      SchedulerApplicationAttempt application); @Override void decUsedResource(String nodeLabel, Resource resourceToDec,
      SchedulerApplicationAttempt application); void incAMUsedResource(String nodeLabel, Resource resourceToInc,
      SchedulerApplicationAttempt application); void decAMUsedResource(String nodeLabel, Resource resourceToDec,
      SchedulerApplicationAttempt application); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); Collection<FiCaSchedulerApp> getPendingApplications(); Collection<FiCaSchedulerApp> getApplications(); Collection<FiCaSchedulerApp> getAllApplications(); Resource getTotalPendingResourcesConsideringUserLimit(
      Resource clusterResources, String partition,
      boolean deductReservedFromPending); @Override void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void attachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); @Override void detachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); Map<String, TreeSet<RMContainer>> getIgnoreExclusivityRMContainers(); void setCapacity(float capacity); void setCapacity(String nodeLabel, float capacity); void setAbsoluteCapacity(float absoluteCapacity); void setAbsoluteCapacity(String nodeLabel, float absoluteCapacity); void setMaxApplicationsPerUser(int maxApplicationsPerUser); void setMaxApplications(int maxApplications); void setMaxAMResourcePerQueuePercent(
      float maxAMResourcePerQueuePercent); OrderingPolicy<FiCaSchedulerApp> getOrderingPolicy(); @Override Priority getDefaultApplicationPriority(); void updateApplicationPriority(SchedulerApplication<FiCaSchedulerApp> app,
      Priority newAppPriority); OrderingPolicy<FiCaSchedulerApp> getPendingAppsOrderingPolicy(); @Override void stopQueue(); Set<String> getAllUsers();  }
@SuppressWarnings({ "unchecked", "rawtypes" }) @Test public void testComputeUserLimitAndSetHeadroom() throws IOException { LeafQueue qb = stubLeafQueue((LeafQueue)queues.get(B)); qb.setMaxCapacity(1.0f); final String user_0 = "user_0"; final String user_1 = "user_1"; String host_0 = "127.0.0.1"; FiCaSchedulerNode node_0 = TestUtils.getMockNode(host_0, DEFAULT_RACK, 0, 8*GB); String host_1 = "127.0.0.2"; FiCaSchedulerNode node_1 = TestUtils.getMockNode(host_1, DEFAULT_RACK, 0, 8*GB); Map<NodeId, FiCaSchedulerNode> nodes = ImmutableMap.of(node_0.getNodeID(), node_0, node_1.getNodeID(), node_1); final int numNodes = 2; Resource clusterResource = Resources.createResource(numNodes * (8*GB), 1); when(csContext.getNumClusterNodes()).thenReturn(numNodes); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); CapacitySchedulerQueueManager mockCapacitySchedulerQueueManager = mock(CapacitySchedulerQueueManager.class); QueueStateManager mockQueueStateManager = mock(QueueStateManager.class); when(mockCapacitySchedulerQueueManager.getQueueStateManager()).thenReturn( mockQueueStateManager); when(csContext.getCapacitySchedulerQueueManager()).thenReturn( mockCapacitySchedulerQueueManager); qb.setUserLimit(100); qb.setUserLimitFactor(1); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); final ApplicationAttemptId appAttemptId_0 = TestUtils.getMockApplicationAttemptId(0, 0); FiCaSchedulerApp app_0 = new FiCaSchedulerApp(appAttemptId_0, user_0, qb, qb.getAbstractUsersManager(), spyRMContext); Map<ApplicationAttemptId, FiCaSchedulerApp> apps = new HashMap<>(); apps.put(app_0.getApplicationAttemptId(), app_0); qb.submitApplicationAttempt(app_0, user_0); Priority u0Priority = TestUtils.createMockPriority(1); SchedulerRequestKey u0SchedKey = toSchedulerKey(u0Priority); app_0.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 4*GB, 1, true, u0Priority, recordFactory))); assertEquals("There should only be 1 active user!", 1, qb.getAbstractUsersManager().getNumActiveUsers()); applyCSAssignment(clusterResource, qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), qb, nodes, apps); qb.computeUserLimitAndSetHeadroom(app_0, clusterResource, "", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null); assertEquals(9*GB,app_0.getHeadroom().getMemorySize()); final ApplicationAttemptId appAttemptId_2 = TestUtils.getMockApplicationAttemptId(2, 0); FiCaSchedulerApp app_2 = new FiCaSchedulerApp(appAttemptId_2, user_1, qb, qb.getAbstractUsersManager(), spyRMContext); apps.put(app_2.getApplicationAttemptId(), app_2); Priority u1Priority = TestUtils.createMockPriority(2); SchedulerRequestKey u1SchedKey = toSchedulerKey(u1Priority); app_2.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 4*GB, 1, true, u1Priority, recordFactory))); qb.submitApplicationAttempt(app_2, user_1); applyCSAssignment(clusterResource, qb.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), qb, nodes, apps); qb.computeUserLimitAndSetHeadroom(app_0, clusterResource, "", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null); assertEquals(8*GB, qb.getUsedResources().getMemorySize()); assertEquals(4*GB, app_0.getCurrentConsumption().getMemorySize()); assertEquals(8*GB, app_0.getHeadroom().getMemorySize()); assertEquals(4*GB, app_2.getCurrentConsumption().getMemorySize()); assertEquals(8*GB, app_2.getHeadroom().getMemorySize()); qb.finishApplication(app_0.getApplicationId(), user_0); qb.finishApplication(app_2.getApplicationId(), user_1); qb.releaseResource(clusterResource, app_0, Resource.newInstance(4*GB, 1), "", null); qb.releaseResource(clusterResource, app_2, Resource.newInstance(4*GB, 1), "", null); qb.setUserLimit(50); qb.setUserLimitFactor(1); final ApplicationAttemptId appAttemptId_1 = TestUtils.getMockApplicationAttemptId(1, 0); FiCaSchedulerApp app_1 = new FiCaSchedulerApp(appAttemptId_1, user_0, qb, qb.getAbstractUsersManager(), spyRMContext); apps.put(app_1.getApplicationAttemptId(), app_1); final ApplicationAttemptId appAttemptId_3 = TestUtils.getMockApplicationAttemptId(3, 0); FiCaSchedulerApp app_3 = new FiCaSchedulerApp(appAttemptId_3, user_1, qb, qb.getAbstractUsersManager(), spyRMContext); apps.put(app_3.getApplicationAttemptId(), app_3); app_1.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 2*GB, 1, true, u0Priority, recordFactory))); app_3.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 2*GB, 1, true, u1Priority, recordFactory))); qb.submitApplicationAttempt(app_1, user_0); qb.submitApplicationAttempt(app_3, user_1); applyCSAssignment(clusterResource, qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), qb, nodes, apps); applyCSAssignment(clusterResource, qb.assignContainers(clusterResource, node_0, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), qb, nodes, apps); qb.computeUserLimitAndSetHeadroom(app_3, clusterResource, "", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null); assertEquals(4*GB, qb.getUsedResources().getMemorySize()); assertEquals(5*GB, app_3.getHeadroom().getMemorySize()); assertEquals(5*GB, app_1.getHeadroom().getMemorySize()); final ApplicationAttemptId appAttemptId_4 = TestUtils.getMockApplicationAttemptId(4, 0); FiCaSchedulerApp app_4 = new FiCaSchedulerApp(appAttemptId_4, user_0, qb, qb.getAbstractUsersManager(), spyRMContext); apps.put(app_4.getApplicationAttemptId(), app_4); qb.submitApplicationAttempt(app_4, user_0); app_4.updateResourceRequests(Collections.singletonList( TestUtils.createResourceRequest(ResourceRequest.ANY, 6*GB, 1, true, u0Priority, recordFactory))); applyCSAssignment(clusterResource, qb.assignContainers(clusterResource, node_1, new ResourceLimits(clusterResource), SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY), qb, nodes, apps); qb.computeUserLimitAndSetHeadroom(app_4, clusterResource, "", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null); qb.computeUserLimitAndSetHeadroom(app_3, clusterResource, "", SchedulingMode.RESPECT_PARTITION_EXCLUSIVITY, null); assertEquals(6*GB, app_3.getHeadroom().getMemorySize()); assertEquals(10*GB, qb.getUsedResources().getMemorySize()); assertEquals(1*GB, app_4.getHeadroom().getMemorySize()); }