@Override
  public GetClusterNodeAttributesResponse getClusterNodeAttributes(
      GetClusterNodeAttributesRequest request)
      throws YarnException, IOException {
    NodeAttributesManager attributesManager =
        rmContext.getNodeAttributesManager();
    Set<NodeAttribute> attributes =
        attributesManager.getClusterNodeAttributes(null);

    GetClusterNodeAttributesResponse response =
        GetClusterNodeAttributesResponse.newInstance(
            attributes.stream().map(attr -> NodeAttributeInfo.newInstance(attr))
                .collect(Collectors.toSet()));
    return response;
  }
@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());
      }
    };
    resourceManager = rm;
    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.
    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);

    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)));
  }