@Test (timeout = 5000)
public void testIsStarvedForFairShare() 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("<weight>.25</weight>");
  out.println("</queue>");
  out.println("<queue name=\"queueB\">");
  out.println("<weight>.75</weight>");
  out.println("</queue>");
  out.println("</allocations>");
  out.close();
  scheduler.reinitialize(conf, resourceManager.getRMContext());
  
  // Add one big node (only care about aggregate capacity)
  RMNode node1 =
      MockNodes.newNodeInfo(1, Resources.createResource(4 * 1024, 4), 1,
          "127.0.0.1");
  NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
  scheduler.handle(nodeEvent1);
  // Queue A wants 3 * 1024. Node update gives this all to A
  createSchedulingRequest(3 * 1024, "queueA", "user1");
  scheduler.update();
  NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
  scheduler.handle(nodeEvent2);
  // Queue B arrives and wants 1 * 1024
  createSchedulingRequest(1 * 1024, "queueB", "user1");
  scheduler.update();
  Collection<FSLeafQueue> queues = scheduler.getQueueManager().getLeafQueues();
  assertEquals(3, queues.size());
  // Queue A should be above fair share, B below.
  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));
    }
  }
  // Node checks in again, should allocate for B
  scheduler.handle(nodeEvent2);
  // B should not be starved for fair share, since entire demand is
  // satisfied.
  for (FSLeafQueue p : queues) {
    if (p.getName().equals("root.queueB")) {
      assertEquals(false, scheduler.isStarvedForFairShare(p));
    }
  }
}