NodeQueueLoadMonitor implements ClusterMonitor { public RMNode selectLocalNode(String hostName, Set<String> blacklist) { if (blacklist.contains(hostName)) { return null; } RMNode node = nodeByHostName.get(hostName); if (node != null) { ClusterNode clusterNode = clusterNodes.get(node.getNodeID()); if (comparator.compareAndIncrement(clusterNode, 1)) { return node; } } return null; } @VisibleForTesting  NodeQueueLoadMonitor(LoadComparator comparator);  NodeQueueLoadMonitor(long nodeComputationInterval,
      LoadComparator comparator, int numNodes); QueueLimitCalculator getThresholdCalculator(); void stop(); void initThresholdCalculator(float sigma, int limitMin, int limitMax); @Override void addNode(List<NMContainerStatus> containerStatuses,
      RMNode rmNode); @Override void removeNode(RMNode removedRMNode); @Override void updateNode(RMNode rmNode); @Override void updateNodeResource(RMNode rmNode, ResourceOption resourceOption); List<NodeId> selectNodes(); List<NodeId> selectLeastLoadedNodes(int k); RMNode selectLocalNode(String hostName, Set<String> blacklist); RMNode selectRackLocalNode(String rackName, Set<String> blacklist); RMNode selectAnyNode(Set<String> blacklist);  }
@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); Set<String> blacklist = new HashSet<>(); RMNode node = selector.selectLocalNode("h1", blacklist); Assert.assertEquals("h1", node.getHostName()); blacklist.add("h1"); node = selector.selectLocalNode("h1", blacklist); Assert.assertNull(node); node = selector.selectLocalNode("h2", blacklist); Assert.assertNull(node); node = selector.selectLocalNode("h3", blacklist); Assert.assertEquals("h3", node.getHostName()); }