ParentQueue extends AbstractCSQueue { @Override public void updateClusterResource(Resource clusterResource, ResourceLimits resourceLimits) { writeLock.lock(); try { Set<String> configuredNodelabels = csContext.getConfiguration() .getConfiguredNodeLabels(getQueuePath()); for (String label : configuredNodelabels) { calculateEffectiveResourcesAndCapacity(label, clusterResource); } for (CSQueue childQueue : childQueues) { ResourceLimits childLimits = getResourceLimitsOfChild(childQueue, clusterResource, resourceLimits.getLimit(), RMNodeLabelsManager.NO_LABEL); childQueue.updateClusterResource(clusterResource, childLimits); } CSQueueUtils.updateQueueStatistics(resourceCalculator, clusterResource, this, labelManager, null); CSQueueUtils.updateConfiguredCapacityMetrics(resourceCalculator, labelManager.getResourceByLabel(null, clusterResource), RMNodeLabelsManager.NO_LABEL, this); } finally { writeLock.unlock(); } } ParentQueue(CapacitySchedulerContext cs,
      String queueName, CSQueue parent, CSQueue old); @Override QueueInfo getQueueInfo(
      boolean includeChildQueues, boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(
      UserGroupInformation user); String toString(); @Override void reinitialize(CSQueue newlyParsedQueue,
      Resource clusterResource); @Override void submitApplication(ApplicationId applicationId, String user,
      String queue); void validateSubmitApplication(ApplicationId applicationId,
      String userName, String queue); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName); @Override void submitApplicationAttempt(FiCaSchedulerApp application,
      String userName, boolean isMoveApp); @Override void finishApplicationAttempt(FiCaSchedulerApp application,
      String queue); @Override void finishApplication(ApplicationId application, String user); @Override CSAssignment assignContainers(Resource clusterResource,
      CandidateNodeSet<FiCaSchedulerNode> candidates,
      ResourceLimits resourceLimits, SchedulingMode schedulingMode); @Override void completedContainer(Resource clusterResource,
      FiCaSchedulerApp application, FiCaSchedulerNode node, 
      RMContainer rmContainer, ContainerStatus containerStatus, 
      RMContainerEventType event, CSQueue completedChildQueue,
      boolean sortQueues); @Override void updateClusterResource(Resource clusterResource,
      ResourceLimits resourceLimits); @Override boolean hasChildQueues(); @Override List<CSQueue> getChildQueues(); @Override void recoverContainer(Resource clusterResource,
      SchedulerApplicationAttempt attempt, RMContainer rmContainer); @Override ActiveUsersManager getAbstractUsersManager(); @Override void collectSchedulerApplications(
      Collection<ApplicationAttemptId> apps); @Override void attachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); @Override void detachContainer(Resource clusterResource,
      FiCaSchedulerApp application, RMContainer rmContainer); int getNumApplications(); void apply(Resource cluster,
      ResourceCommitRequest<FiCaSchedulerApp, FiCaSchedulerNode> request); @Override void stopQueue(); QueueOrderingPolicy getQueueOrderingPolicy();  }
@Test public void testAbsoluteResourceWithChangeInClusterResource() throws Exception { setupSingleLevelQueuesWithAbsoluteResource(csConf); CSQueueStore queues = new CSQueueStore(); CSQueue root = CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); final int memoryPerNode = 10; int coresPerNode = 16; int numNodes = 2; Resource clusterResource = Resources.createResource( numNodes * (memoryPerNode * GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); LeafQueue a = (LeafQueue) queues.get(A); LeafQueue b = (LeafQueue) queues.get(B); assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_A_RESOURCE); assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_B_RESOURCE); assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_A_RESOURCE); assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_B_RESOURCE); numNodes = 1; clusterResource = Resources.createResource(numNodes * (memoryPerNode * GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); Resource QUEUE_B_RESOURCE_HALF = Resource.newInstance(7 * 1024, 11); Resource QUEUE_A_RESOURCE_HALF = Resource.newInstance(3 * 1024, 5); assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_A_RESOURCE); assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_B_RESOURCE); assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_A_RESOURCE_HALF); assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_B_RESOURCE_HALF); coresPerNode = 40; clusterResource = Resources.createResource(numNodes * (memoryPerNode * GB), numNodes * coresPerNode); when(csContext.getNumClusterNodes()).thenReturn(numNodes); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); Resource QUEUE_B_RESOURCE_70PERC = Resource.newInstance(7 * 1024, 27); Resource QUEUE_A_RESOURCE_30PERC = Resource.newInstance(3 * 1024, 12); assertEquals(a.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_A_RESOURCE); assertEquals(b.getQueueResourceQuotas().getConfiguredMinResource(), QUEUE_B_RESOURCE); assertEquals(a.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_A_RESOURCE_30PERC); assertEquals(b.getQueueResourceQuotas().getEffectiveMinResource(), QUEUE_B_RESOURCE_70PERC); }