public RMNode selectLocalNode(
      String hostName, Set<String> blacklist, Resource request) {
    if (blacklist.contains(hostName)) {
      return null;
    }
    RMNode node = nodeByHostName.get(hostName);
    if (node != null) {
      ClusterNode clusterNode = clusterNodes.get(node.getNodeID());
      if (clusterNode != null && comparator
          .compareAndIncrement(clusterNode, 1, request)) {
        return node;
      }
    }
    return null;
  }
@Test
  public void testSelectLocalNode() {
    NodeQueueLoadMonitor selector = new NodeQueueLoadMonitor(
        NodeQueueLoadMonitor.LoadComparator.QUEUE_LENGTH);

    RMNode h1 = createRMNode("h1", 1, -1, 2, 5);
    RMNode h2 = createRMNode("h2", 2, -1, 5, 5);
    RMNode h3 = createRMNode("h3", 3, -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.selectLocalNode(
        "h1", blacklist, defaultResourceRequested);
    Assert.assertEquals("h1", node.getHostName());

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

    node = selector.selectLocalNode(
        "h2", blacklist, defaultResourceRequested);
    Assert.assertNull(node);

    node = selector.selectLocalNode(
        "h3", blacklist, defaultResourceRequested);
    Assert.assertEquals("h3", node.getHostName());
  }