@Test(timeout = 120000)
public void testGetClusterNodeAttributes() throws IOException, YarnException {

  Configuration newConf = NodeAttributeTestUtils.getRandomDirConf(null);
  MockRM rm = new MockRM(newConf) {
    protected ClientRMService createClientRMService() {
      return new ClientRMService(this.rmContext, scheduler, this.rmAppManager,
          this.applicationACLsManager, this.queueACLsManager,
          this.getRMContext().getRMDelegationTokenSecretManager());
    }
  };
  rm.start();
  NodeAttributesManager mgr = rm.getRMContext().getNodeAttributesManager();
  NodeId host1 = NodeId.newInstance("host1", 0);
  NodeId host2 = NodeId.newInstance("host2", 0);
  NodeAttribute gpu = NodeAttribute
      .newInstance(NodeAttribute.PREFIX_CENTRALIZED, "GPU",
          NodeAttributeType.STRING, "nvida");
  NodeAttribute os = NodeAttribute
      .newInstance(NodeAttribute.PREFIX_CENTRALIZED, "OS",
          NodeAttributeType.STRING, "windows64");
  NodeAttribute docker = NodeAttribute
      .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "DOCKER",
          NodeAttributeType.STRING, "docker0");
  Map<String, Set<NodeAttribute>> nodes = new HashMap<>();
  nodes.put(host1.getHost(), ImmutableSet.of(gpu, os));
  nodes.put(host2.getHost(), ImmutableSet.of(docker));
  mgr.addNodeAttributes(nodes);
  // 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);
  GetClusterNodeAttributesRequest request =
      GetClusterNodeAttributesRequest.newInstance();
  GetClusterNodeAttributesResponse response =
      client.getClusterNodeAttributes(request);
  Set<NodeAttributeInfo> attributes = response.getNodeAttributes();
  Assert.assertEquals("Size not correct", 3, attributes.size());
  Assert.assertTrue(attributes.contains(NodeAttributeInfo.newInstance(gpu)));
  Assert.assertTrue(attributes.contains(NodeAttributeInfo.newInstance(os)));
  Assert
      .assertTrue(attributes.contains(NodeAttributeInfo.newInstance(docker)));
  rpc.stopProxy(client, conf);
  rm.close();
}