ParentQueue extends AbstractCSQueue { private void deriveCapacityFromAbsoluteConfigurations(String label, Resource clusterResource, ResourceCalculator rc, CSQueue childQueue) { childQueue.getQueueCapacities().setCapacity(label, rc.divide(clusterResource, childQueue.getQueueResourceQuotas().getEffectiveMinResource(label), getQueueResourceQuotas().getEffectiveMinResource(label))); childQueue.getQueueCapacities().setMaximumCapacity(label, rc.divide(clusterResource, childQueue.getQueueResourceQuotas().getEffectiveMaxResource(label), getQueueResourceQuotas().getEffectiveMaxResource(label))); childQueue.getQueueCapacities().setAbsoluteCapacity(label, childQueue.getQueueCapacities().getCapacity(label) * getQueueCapacities().getAbsoluteCapacity(label)); childQueue.getQueueCapacities().setAbsoluteMaximumCapacity(label, childQueue.getQueueCapacities().getMaximumCapacity(label) * getQueueCapacities().getAbsoluteMaximumCapacity(label)); if (childQueue instanceof LeafQueue) { LeafQueue leafQueue = (LeafQueue) childQueue; CapacitySchedulerConfiguration conf = csContext.getConfiguration(); int maxApplications = conf.getMaximumApplicationsPerQueue(childQueue.getQueuePath()); if (maxApplications < 0) { int maxGlobalPerQueueApps = conf.getGlobalMaximumApplicationsPerQueue(); if (maxGlobalPerQueueApps > 0) { maxApplications = (int) (maxGlobalPerQueueApps * childQueue.getQueueCapacities().getAbsoluteCapacity(label)); } else { maxApplications = (int) (conf.getMaximumSystemApplications() * childQueue.getQueueCapacities().getAbsoluteCapacity(label)); } } leafQueue.setMaxApplications(maxApplications); int maxApplicationsPerUser = Math.min(maxApplications, (int) (maxApplications * (leafQueue.getUsersManager().getUserLimit() / 100.0f) * leafQueue.getUsersManager().getUserLimitFactor())); leafQueue.setMaxApplicationsPerUser(maxApplicationsPerUser); LOG.info("LeafQueue:" + leafQueue.getQueuePath() + ", maxApplications=" + maxApplications + ", maxApplicationsPerUser=" + maxApplicationsPerUser + ", Abs Cap:" + childQueue.getQueueCapacities().getAbsoluteCapacity(label) + ", Cap: " + childQueue.getQueueCapacities().getCapacity(label) + ", MaxCap : " + childQueue.getQueueCapacities().getMaximumCapacity(label)); } } 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 testDeriveCapacityFromAbsoluteConfigurations() throws Exception { setupSingleLevelQueuesWithAbsoluteResource(csConf); CSQueueStore queues = new CSQueueStore(); CSQueue root = CapacitySchedulerQueueManager.parseQueue(csContext, csConf, null, CapacitySchedulerConfiguration.ROOT, queues, queues, TestUtils.spyHook); int numNodes = 2; final long memoryPerNode = (QUEUE_A_RESOURCE.getMemorySize() + QUEUE_B_RESOURCE.getMemorySize()) / numNodes; int coresPerNode = (QUEUE_A_RESOURCE.getVirtualCores() + QUEUE_B_RESOURCE.getVirtualCores()) / numNodes; Resource clusterResource = Resources.createResource( numNodes * memoryPerNode, 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); float queueAScale = (float) QUEUE_A_RESOURCE.getMemorySize() / (float) clusterResource.getMemorySize(); float queueBScale = (float) QUEUE_B_RESOURCE.getMemorySize() / (float) clusterResource.getMemorySize(); assertEquals(queueAScale, a.getQueueCapacities().getCapacity(), DELTA); assertEquals(1f, a.getQueueCapacities().getMaximumCapacity(), DELTA); assertEquals(queueAScale, a.getQueueCapacities().getAbsoluteCapacity(), DELTA); assertEquals(1f, a.getQueueCapacities().getAbsoluteMaximumCapacity(), DELTA); assertEquals((int) (csConf.getMaximumSystemApplications() * queueAScale), a.getMaxApplications()); assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser()); assertEquals(queueBScale, b.getQueueCapacities().getCapacity(), DELTA); assertEquals(1f, b.getQueueCapacities().getMaximumCapacity(), DELTA); assertEquals(queueBScale, b.getQueueCapacities().getAbsoluteCapacity(), DELTA); assertEquals(1f, b.getQueueCapacities().getAbsoluteMaximumCapacity(), DELTA); assertEquals((int) (csConf.getMaximumSystemApplications() * queueBScale), b.getMaxApplications()); assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser()); csConf.setGlobalMaximumApplicationsPerQueue(20000); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); assertEquals((int) (csConf.getGlobalMaximumApplicationsPerQueue() * queueAScale), a.getMaxApplications()); assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser()); assertEquals((int) (csConf.getGlobalMaximumApplicationsPerQueue() * queueBScale), b.getMaxApplications()); assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser()); int queueAMaxApplications = 30000; int queueBMaxApplications = 30000; csConf.set("yarn.scheduler.capacity." + Q_A + ".maximum-applications", Integer.toString(queueAMaxApplications)); csConf.set("yarn.scheduler.capacity." + Q_B + ".maximum-applications", Integer.toString(queueBMaxApplications)); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); assertEquals(queueAMaxApplications, a.getMaxApplications()); assertEquals(a.getMaxApplications(), a.getMaxApplicationsPerUser()); assertEquals(queueBMaxApplications, b.getMaxApplications()); assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser()); int halfPercent = 50; int oneAndQuarterPercent = 125; a.getUsersManager().setUserLimit(halfPercent); b.getUsersManager().setUserLimit(oneAndQuarterPercent); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); assertEquals(a.getMaxApplications() * halfPercent / 100, a.getMaxApplicationsPerUser()); assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser()); float userLimitFactorQueueA = 0.9f; float userLimitFactorQueueB = 1.1f; a.getUsersManager().setUserLimit(halfPercent); a.getUsersManager().setUserLimitFactor(userLimitFactorQueueA); b.getUsersManager().setUserLimit(100); b.getUsersManager().setUserLimitFactor(userLimitFactorQueueB); root.updateClusterResource(clusterResource, new ResourceLimits(clusterResource)); assertEquals((int) (a.getMaxApplications() * halfPercent * userLimitFactorQueueA / 100), a.getMaxApplicationsPerUser()); assertEquals(b.getMaxApplications(), b.getMaxApplicationsPerUser()); }