public RMNode selectAnyNode(Set<String> blacklist, Resource request) {
    List<NodeId> nodeIds = getCandidatesForSelectAnyNode();
    int size = nodeIds.size();
    if (size <= 0) {
      return null;
    }
    Random rand = new Random();
    int startIndex = rand.nextInt(size);
    for (int i = 0; i < size; ++i) {
      int index = i + startIndex;
      index %= size;
      NodeId nodeId = nodeIds.get(index);
      if (nodeId != null && !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 testSelectAnyNode() {
    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, 10);

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

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

    selector.computeTask.run();

    Assert.assertEquals(2, selector.getSortedNodes().size());

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

    // if node has been added to blacklist
    blacklist.add("h1");
    node = selector.selectAnyNode(blacklist, defaultResourceRequested);
    Assert.assertEquals("h3", node.getHostName());

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