public RMNode selectRackLocalNode(
      String rackName, Set<String> blacklist, Resource request) {
    Set<NodeId> nodesOnRack = nodeIdsByRack.get(rackName);
    if (nodesOnRack != null) {
      for (NodeId nodeId : nodesOnRack) {
        if (!blacklist.contains(nodeId.getHost())) {
          ClusterNode node = clusterNodes.get(nodeId);
          if (node != null &&
              comparator.compareAndIncrement(node, 1, request)) {
            return nodeByHostName.get(nodeId.getHost());
          }
        }
      }
    }
    return null;
  }
@Test
  public void testSelectRackLocalNode() {
    NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor(
        NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH);

    RMNode h1 = createRMNode("h1", 1, "rack1", -1, 2, 5);
    RMNode h2 = createRMNode("h2", 2, "rack2", -1, 5, 5);
    RMNode h3 = createRMNode("h3", 3, "rack2", -1, 4, 5);

    selector.addNode(null, h1);
    selector.addNode(null, h2);
    selector.addNode(null, h3);

    selector.updateNode(h1);
    selector.updateNode(h2);
    selector.updateNode(h3);

    // basic test for selecting node which has queue length less
    // than queue capacity.
    Set<String> blacklist = new HashSet<>();
    RMNode node = selector.selectRackLocalNode(
        "rack1", blacklist, defaultResourceRequested);
    Assert.assertEquals("h1", node.getHostName());

    // if node has been added to blacklist
    blacklist.add("h1");
    node = selector.selectRackLocalNode(
        "rack1", blacklist, defaultResourceRequested);
    Assert.assertNull(node);

    node = selector.selectRackLocalNode(
        "rack2", blacklist, defaultResourceRequested);
    Assert.assertEquals("h3", node.getHostName());

    blacklist.add("h3");
    node = selector.selectRackLocalNode(
        "rack2", blacklist, defaultResourceRequested);
    Assert.assertNull(node);
  }