@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>.2</weight>");
  out.println("</queue>");
  out.println("<queue name=\"queueB\">");
  out.println("<weight>.8</weight>");
  out.println("<fairSharePreemptionThreshold>.4</fairSharePreemptionThreshold>");
  out.println("<queue name=\"queueB1\">");
  out.println("</queue>");
  out.println("<queue name=\"queueB2\">");
  out.println("<fairSharePreemptionThreshold>.6</fairSharePreemptionThreshold>");
  out.println("</queue>");
  out.println("</queue>");
  out.println("<defaultFairSharePreemptionThreshold>.5</defaultFairSharePreemptionThreshold>");
  out.println("</allocations>");
  out.close();
  resourceManager = new MockRM(conf);
  resourceManager.start();
  scheduler = (FairScheduler) resourceManager.getResourceScheduler();
  // Add one big node (only care about aggregate capacity)
  RMNode node1 =
      MockNodes.newNodeInfo(1, Resources.createResource(10 * 1024, 10), 1,
          "127.0.0.1");
  NodeAddedSchedulerEvent nodeEvent1 = new NodeAddedSchedulerEvent(node1);
  scheduler.handle(nodeEvent1);
  scheduler.update();
  // Queue A wants 4 * 1024. Node update gives this all to A
  createSchedulingRequest(1 * 1024, "queueA", "user1", 4);
  scheduler.update();
  NodeUpdateSchedulerEvent nodeEvent2 = new NodeUpdateSchedulerEvent(node1);
  for (int i = 0; i < 4; i ++) {
    scheduler.handle(nodeEvent2);
  }
  QueueManager queueMgr = scheduler.getQueueManager();
  FSLeafQueue queueA = queueMgr.getLeafQueue("queueA", false);
  assertEquals(4 * 1024, queueA.getResourceUsage().getMemory());
  // Both queue B1 and queue B2 want 3 * 1024
  createSchedulingRequest(1 * 1024, "queueB.queueB1", "user1", 3);
  createSchedulingRequest(1 * 1024, "queueB.queueB2", "user1", 3);
  scheduler.update();
  for (int i = 0; i < 4; i ++) {
    scheduler.handle(nodeEvent2);
  }
  FSLeafQueue queueB1 = queueMgr.getLeafQueue("queueB.queueB1", false);
  FSLeafQueue queueB2 = queueMgr.getLeafQueue("queueB.queueB2", false);
  assertEquals(2 * 1024, queueB1.getResourceUsage().getMemory());
  assertEquals(2 * 1024, queueB2.getResourceUsage().getMemory());
  // For queue B1, the fairSharePreemptionThreshold is 0.4, and the fair share
  // threshold is 1.6 * 1024
  assertFalse(queueB1.isStarvedForFairShare());
  // For queue B2, the fairSharePreemptionThreshold is 0.6, and the fair share
  // threshold is 2.4 * 1024
  assertTrue(queueB2.isStarvedForFairShare());
  // Node checks in again
  scheduler.handle(nodeEvent2);
  scheduler.handle(nodeEvent2);
  assertEquals(3 * 1024, queueB1.getResourceUsage().getMemory());
  assertEquals(3 * 1024, queueB2.getResourceUsage().getMemory());
  // Both queue B1 and queue B2 usages go to 3 * 1024
  assertFalse(queueB1.isStarvedForFairShare());
  assertFalse(queueB2.isStarvedForFairShare());
}