@Override
  public GetNodesToAttributesResponse getNodesToAttributes(
      GetNodesToAttributesRequest request) throws YarnException, IOException {
    NodeAttributesManager attributesManager =
        rmContext.getNodeAttributesManager();
    GetNodesToAttributesResponse response = GetNodesToAttributesResponse
        .newInstance(
            attributesManager.getNodesToAttributes(request.getHostNames()));
    return response;
  }
@Test(timeout = 120000)
  public void testGetNodesToAttributes() 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());
      }
    };
    resourceManager = rm;
    rm.start();

    NodeAttributesManager mgr = rm.getRMContext().getNodeAttributesManager();
    String node1 = "host1";
    String node2 = "host2";
    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");
    NodeAttribute dist = NodeAttribute
        .newInstance(NodeAttribute.PREFIX_DISTRIBUTED, "VERSION",
            NodeAttributeType.STRING, "3_0_2");
    Map<String, Set<NodeAttribute>> nodes = new HashMap<>();
    nodes.put(node1, ImmutableSet.of(gpu, os, dist));
    nodes.put(node2, ImmutableSet.of(docker, dist));
    mgr.addNodeAttributes(nodes);
    // Create a client.
    conf = new Configuration();
    rpc = YarnRPC.create(conf);
    InetSocketAddress rmAddress = rm.getClientRMService().getBindAddress();
    LOG.info("Connecting to ResourceManager at " + rmAddress);
    client = (ApplicationClientProtocol) rpc.getProxy(
        ApplicationClientProtocol.class, rmAddress, conf);

    // Specify null for hostnames.
    GetNodesToAttributesRequest request1 =
        GetNodesToAttributesRequest.newInstance(null);
    GetNodesToAttributesResponse response1 =
        client.getNodesToAttributes(request1);
    Map<String, Set<NodeAttribute>> hostToAttrs =
        response1.getNodeToAttributes();
    Assert.assertEquals(2, hostToAttrs.size());

    Assert.assertTrue(hostToAttrs.get(node2).contains(dist));
    Assert.assertTrue(hostToAttrs.get(node2).contains(docker));
    Assert.assertTrue(hostToAttrs.get(node1).contains(dist));

    // Specify particular node
    GetNodesToAttributesRequest request2 =
        GetNodesToAttributesRequest.newInstance(ImmutableSet.of(node1));
    GetNodesToAttributesResponse response2 =
        client.getNodesToAttributes(request2);
    hostToAttrs = response2.getNodeToAttributes();
    Assert.assertEquals(1, response2.getNodeToAttributes().size());
    Assert.assertTrue(hostToAttrs.get(node1).contains(dist));

    // Test queury with empty set
    GetNodesToAttributesRequest request3 =
        GetNodesToAttributesRequest.newInstance(Collections.emptySet());
    GetNodesToAttributesResponse response3 =
        client.getNodesToAttributes(request3);
    hostToAttrs = response3.getNodeToAttributes();
    Assert.assertEquals(2, hostToAttrs.size());

    Assert.assertTrue(hostToAttrs.get(node2).contains(dist));
    Assert.assertTrue(hostToAttrs.get(node2).contains(docker));
    Assert.assertTrue(hostToAttrs.get(node1).contains(dist));

    // test invalid hostname
    GetNodesToAttributesRequest request4 =
        GetNodesToAttributesRequest.newInstance(ImmutableSet.of("invalid"));
    GetNodesToAttributesResponse response4 =
        client.getNodesToAttributes(request4);
    hostToAttrs = response4.getNodeToAttributes();
    Assert.assertEquals(0, hostToAttrs.size());
  }