### Question:
FairScheduler implements ResourceScheduler { boolean isStarvedForFairShare(FSLeafQueue sched) { Resource desiredFairShare = Resources.max( Resources.multiply(sched.getFairShare(), .5), sched.getDemand()); return Resources.lessThan(sched.getResourceUsage(), desiredFairShare); } FairScheduler(); FairSchedulerConfiguration getConf(); QueueManager getQueueManager(); RMContainerTokenSecretManager getContainerTokenSecretManager(); synchronized double getAppWeight(AppSchedulable app); @Override Resource getMinimumResourceCapability(); @Override Resource getMaximumResourceCapability(); double getNodeLocalityThreshold(); double getRackLocalityThreshold(); Resource getClusterCapacity(); Clock getClock(); FairSchedulerEventLog getEventLog(); @Override Allocation allocate(ApplicationAttemptId appAttemptId,
      List<ResourceRequest> ask, List<ContainerId> release); @Override SchedulerNodeReport getNodeReport(NodeId nodeId); FSSchedulerApp getSchedulerApp(ApplicationAttemptId appAttemptId); @Override SchedulerAppReport getSchedulerAppInfo(
      ApplicationAttemptId appAttemptId); @Override QueueMetrics getRootQueueMetrics(); @Override void handle(SchedulerEvent event); @Override void recover(RMState state); @Override synchronized void reinitialize(Configuration conf, RMContext rmContext); @Override QueueInfo getQueueInfo(String queueName, boolean includeChildQueues,
      boolean recursive); @Override List<QueueUserACLInfo> getQueueUserAclInfo(); @Override int getNumClusterNodes(); static final Resource CONTAINER_RESERVED; }### Answer:
@Test public void testIsStarvedForFairShare() throws Exception { Configuration conf = createConfiguration(); conf.set(FairSchedulerConfiguration.ALLOCATION_FILE, ALLOC_FILE); scheduler.reinitialize(conf, resourceManager.getRMContext()); PrintWriter out = new PrintWriter(new FileWriter(ALLOC_FILE)); out.println("<?xml version=\"1.0\"?>"); out.println("<allocations>"); out.println("<queue name=\"queueA\">"); out.println("<weight>.25</weight>"); out.println("</queue>"); out.println("<queue name=\"queueB\">"); out.println("<weight>.75</weight>"); out.println("</queue>"); out.println("</allocations>"); out.close(); QueueManager queueManager = scheduler.getQueueManager(); queueManager.initialize(); RMNode node1 = MockNodes.newNodeInfo(1, Resources.createResource(4 * 1024)); NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1); scheduler.handle(nodeEvent1); createSchedulingRequest(3 * 1024, "queueA", "user1"); scheduler.update(); NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1, new LinkedList<ContainerStatus>(), new LinkedList<ContainerStatus>()); 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.isStarvedForFairShare(p)); } else if (p.getName().equals("root.queueB")) { assertEquals(true, scheduler.isStarvedForFairShare(p)); } } scheduler.handle(nodeEvent2); for (FSLeafQueue p : queues) { if (p.getName().equals("root.queueB")) { assertEquals(false, scheduler.isStarvedForFairShare(p)); } } }