@Test
public void testGetClusterNodes() throws Exception {

  MockRM rm = new MockRM() {
    protected ClientRMService createClientRMService() {
      return new ClientRMService(this.rmContext, scheduler,
        this.rmAppManager, this.applicationACLsManager, this.queueACLsManager,
        this.getRMContext().getRMDelegationTokenSecretManager());
    };
  };
  rm.start();
  RMNodeLabelsManager labelsMgr = rm.getRMContext().getNodeLabelManager();
  labelsMgr.addToCluserNodeLabelsWithDefaultExclusivity(ImmutableSet.of("x", "y"));
  // Add a healthy node with label = x
  MockNM node = rm.registerNode("host1:1234", 1024);
  Map<NodeId, Set<String>> map = new HashMap<NodeId, Set<String>>();
  map.put(node.getNodeId(), ImmutableSet.of("x"));
  labelsMgr.replaceLabelsOnNode(map);
  rm.sendNodeStarted(node);
  node.nodeHeartbeat(true);
  
  // Add and lose a node with label = y
  MockNM lostNode = rm.registerNode("host2:1235", 1024);
  rm.sendNodeStarted(lostNode);
  lostNode.nodeHeartbeat(true);
  rm.waitForState(lostNode.getNodeId(), NodeState.RUNNING);
  rm.sendNodeLost(lostNode);
  // Create a client.
  Configuration conf = new Configuration();
  YarnRPC rpc = YarnRPC.create(conf);
  InetSocketAddress rmAddress = rm.getClientRMService().getBindAddress();
  LOG.info("Connecting to ResourceManager at " + rmAddress);
  ApplicationClientProtocol client =
      (ApplicationClientProtocol) rpc
        .getProxy(ApplicationClientProtocol.class, rmAddress, conf);
  // Make call
  GetClusterNodesRequest request =
      GetClusterNodesRequest.newInstance(EnumSet.of(NodeState.RUNNING));
  List<NodeReport> nodeReports =
      client.getClusterNodes(request).getNodeReports();
  Assert.assertEquals(1, nodeReports.size());
  Assert.assertNotSame("Node is expected to be healthy!", NodeState.UNHEALTHY,
      nodeReports.get(0).getNodeState());
  
  // Check node's label = x
  Assert.assertTrue(nodeReports.get(0).getNodeLabels().contains("x"));
  Assert.assertNull(nodeReports.get(0).getDecommissioningTimeout());
  Assert.assertNull(nodeReports.get(0).getNodeUpdateType());
  // Now make the node unhealthy.
  node.nodeHeartbeat(false);
  rm.waitForState(node.getNodeId(), NodeState.UNHEALTHY);
  // Call again
  nodeReports = client.getClusterNodes(request).getNodeReports();
  Assert.assertEquals("Unhealthy nodes should not show up by default", 0,
      nodeReports.size());
  
  // Change label of host1 to y
  map = new HashMap<NodeId, Set<String>>();
  map.put(node.getNodeId(), ImmutableSet.of("y"));
  labelsMgr.replaceLabelsOnNode(map);
  
  // Now query for UNHEALTHY nodes
  request = GetClusterNodesRequest.newInstance(EnumSet.of(NodeState.UNHEALTHY));
  nodeReports = client.getClusterNodes(request).getNodeReports();
  Assert.assertEquals(1, nodeReports.size());
  Assert.assertEquals("Node is expected to be unhealthy!", NodeState.UNHEALTHY,
      nodeReports.get(0).getNodeState());
  
  Assert.assertTrue(nodeReports.get(0).getNodeLabels().contains("y"));
  Assert.assertNull(nodeReports.get(0).getDecommissioningTimeout());
  Assert.assertNull(nodeReports.get(0).getNodeUpdateType());
  
  // Remove labels of host1
  map = new HashMap<NodeId, Set<String>>();
  map.put(node.getNodeId(), ImmutableSet.of("y"));
  labelsMgr.removeLabelsFromNode(map);
  
  // Query all states should return all nodes
  rm.registerNode("host3:1236", 1024);
  request = GetClusterNodesRequest.newInstance(EnumSet.allOf(NodeState.class));
  nodeReports = client.getClusterNodes(request).getNodeReports();
  Assert.assertEquals(3, nodeReports.size());
  
  // All host1-3's label should be empty (instead of null)
  for (NodeReport report : nodeReports) {
    Assert.assertTrue(report.getNodeLabels() != null
        && report.getNodeLabels().isEmpty());
    Assert.assertNull(report.getDecommissioningTimeout());
    Assert.assertNull(report.getNodeUpdateType());
  }
  rpc.stopProxy(client, conf);
  rm.close();
}