@Test (timeout = 5000) public void testIsStarvedForMinShare() throws Exception { conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE); PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE)); out.println("<?xml version=\"1.0\"?>"); out.println("<allocations>"); out.println("<queue name=\"queueA\">"); out.println("<minResources>2048mb,0vcores</minResources>"); out.println("</queue>"); out.println("<queue name=\"queueB\">"); out.println("<minResources>2048mb,0vcores</minResources>"); out.println("</queue>"); out.println("</allocations>"); out.close(); scheduler.init(conf); scheduler.start(); scheduler.reinitialize(conf, resourceManager.getRMContext()); RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(4 * 1024, 4), 1, "127.0.0.1"); NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1); scheduler.handle(nodeEvent1); createSchedulingRequest(3 * 1024, "queueA", "user1"); scheduler.update(); NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1); scheduler.handle(nodeEvent2); createSchedulingRequest(1 * 1024, "queueB", "user1"); scheduler.update(); Collection<FSLeafQueue> queues = scheduler.getQueueManager().getLeafQueues(); assertEquals(3, queues.size()); for (FSLeafQueue p : queues) { if (p.getName().equals("root.queueA")) { assertEquals(false, scheduler.isStarvedForMinShare(p)); } else if (p.getName().equals("root.queueB")) { assertEquals(true, scheduler.isStarvedForMinShare(p)); } } scheduler.handle(nodeEvent2); for (FSLeafQueue p : queues) { if (p.getName().equals("root.queueB")) { assertEquals(false, scheduler.isStarvedForMinShare(p)); } } }
boolean isStarvedForMinShare(FSLeafQueue sched) { Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterResource, sched.getMinShare(), sched.getDemand()); return Resources.lessThan(RESOURCE_CALCULATOR, clusterResource, sched.getResourceUsage(), desiredShare); }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { boolean isStarvedForMinShare(FSLeafQueue sched) { Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterResource, sched.getMinShare(), sched.getDemand()); return Resources.lessThan(RESOURCE_CALCULATOR, clusterResource, sched.getResourceUsage(), desiredShare); } }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { boolean isStarvedForMinShare(FSLeafQueue sched) { Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterResource, sched.getMinShare(), sched.getDemand()); return Resources.lessThan(RESOURCE_CALCULATOR, clusterResource, sched.getResourceUsage(), desiredShare); } FairScheduler(); }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { boolean isStarvedForMinShare(FSLeafQueue sched) { Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterResource, sched.getMinShare(), sched.getDemand()); return Resources.lessThan(RESOURCE_CALCULATOR, clusterResource, sched.getResourceUsage(), desiredShare); } FairScheduler(); FairSchedulerConfiguration getConf(); QueueManager getQueueManager(); synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); synchronized ResourceWeights getAppWeight(FSAppAttempt app); Resource getIncrementResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); long getNodeLocalityDelayMs(); long getRackLocalityDelayMs(); boolean isContinuousSchedulingEnabled(); synchronized int getContinuousSchedulingSleepMs(); synchronized Clock getClock(); FairSchedulerEventLog getEventLog(); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals); FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); AllocationConfiguration getAllocationConfiguration(); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized String moveApplication(ApplicationId appId,
      String queueName); }
FairScheduler extends
    AbstractYarnScheduler<FSAppAttempt, FSSchedulerNode> { boolean isStarvedForMinShare(FSLeafQueue sched) { Resource desiredShare = Resources.min(RESOURCE_CALCULATOR, clusterResource, sched.getMinShare(), sched.getDemand()); return Resources.lessThan(RESOURCE_CALCULATOR, clusterResource, sched.getResourceUsage(), desiredShare); } FairScheduler(); FairSchedulerConfiguration getConf(); QueueManager getQueueManager(); synchronized RMContainerTokenSecretManager getContainerTokenSecretManager(); synchronized ResourceWeights getAppWeight(FSAppAttempt app); Resource getIncrementResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); long getNodeLocalityDelayMs(); long getRackLocalityDelayMs(); boolean isContinuousSchedulingEnabled(); synchronized int getContinuousSchedulingSleepMs(); synchronized Clock getClock(); FairSchedulerEventLog getEventLog(); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release, List<String> blacklistAdditions, List<String> blacklistRemovals); FSAppAttempt getSchedulerApp(ApplicationAttemptId appAttemptId); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); synchronized void setRMContext(RMContext rmContext); @Override void serviceInit(Configuration conf); @Override void serviceStart(); @Override void serviceStop(); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); @Override synchronized boolean checkAccess(UserGroupInformation callerUGI,
      QueueACL acl, String queueName); AllocationConfiguration getAllocationConfiguration(); @Override List<ApplicationAttemptId> getAppsInQueue(String queueName); @Override synchronized String moveApplication(ApplicationId appId,
      String queueName); static final Resource CONTAINER_RESERVED; }