ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetNodesToAttributesResponse getNodesToAttributes( GetNodesToAttributesRequest request) throws YarnException, IOException { NodeAttributesManager attributesManager = rmContext.getNodeAttributesManager(); GetNodesToAttributesResponse response = GetNodesToAttributesResponse .newInstance( attributesManager.getNodesToAttributes(request.getHostNames())); return response; } ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager);  ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager, Clock clock); @Private InetSocketAddress getBindAddress(); @Override GetNewApplicationResponse getNewApplication(
      GetNewApplicationRequest request); @Override GetApplicationReportResponse getApplicationReport(
      GetApplicationReportRequest request); @Override GetApplicationAttemptReportResponse getApplicationAttemptReport(
      GetApplicationAttemptReportRequest request); @Override GetApplicationAttemptsResponse getApplicationAttempts(
      GetApplicationAttemptsRequest request); @Override GetContainerReportResponse getContainerReport(
      GetContainerReportRequest request); @Override GetContainersResponse getContainers(GetContainersRequest request); @Override SubmitApplicationResponse submitApplication(
      SubmitApplicationRequest request); @SuppressWarnings("unchecked") @Override FailApplicationAttemptResponse failApplicationAttempt(
      FailApplicationAttemptRequest request); @SuppressWarnings("unchecked") @Override KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetApplicationsResponse getApplications(GetApplicationsRequest request); @Override GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request); @Override GetQueueInfoResponse getQueueInfo(GetQueueInfoRequest request); @Override GetQueueUserAclsInfoResponse getQueueUserAcls(
      GetQueueUserAclsInfoRequest request); @Override GetDelegationTokenResponse getDelegationToken(
      GetDelegationTokenRequest request); @Override RenewDelegationTokenResponse renewDelegationToken(
      RenewDelegationTokenRequest request); @Override CancelDelegationTokenResponse cancelDelegationToken(
      CancelDelegationTokenRequest request); @SuppressWarnings("unchecked") @Override MoveApplicationAcrossQueuesResponse moveApplicationAcrossQueues(
      MoveApplicationAcrossQueuesRequest request); @VisibleForTesting Server getServer(); @Override GetNewReservationResponse getNewReservation(
      GetNewReservationRequest request); @Override ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override ReservationListResponse listReservations(
        ReservationListRequest requestInfo); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetLabelsToNodesResponse getLabelsToNodes(
      GetLabelsToNodesRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request); @Override UpdateApplicationPriorityResponse updateApplicationPriority(
      UpdateApplicationPriorityRequest request); @SuppressWarnings("unchecked") @Override SignalContainerResponse signalToContainer(
      SignalContainerRequest request); @Override UpdateApplicationTimeoutsResponse updateApplicationTimeouts(
      UpdateApplicationTimeoutsRequest request); @Override GetAllResourceProfilesResponse getResourceProfiles(
      GetAllResourceProfilesRequest request); @Override GetResourceProfileResponse getResourceProfile(
      GetResourceProfileRequest request); @Override GetAllResourceTypeInfoResponse getResourceTypeInfo(
      GetAllResourceTypeInfoRequest request); @Override GetAttributesToNodesResponse getAttributesToNodes(
      GetAttributesToNodesRequest request); @Override GetClusterNodeAttributesResponse getClusterNodeAttributes(
      GetClusterNodeAttributesRequest request); @Override GetNodesToAttributesResponse getNodesToAttributes(
      GetNodesToAttributesRequest request); @VisibleForTesting void setDisplayPerUserApps(boolean displayPerUserApps);  }
@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()); } }; 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); 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); 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)); 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)); 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)); GetNodesToAttributesRequest request4 = GetNodesToAttributesRequest.newInstance(ImmutableSet.of("invalid")); GetNodesToAttributesResponse response4 = client.getNodesToAttributes(request4); hostToAttrs = response4.getNodeToAttributes(); Assert.assertEquals(0, hostToAttrs.size()); rpc.stopProxy(client, conf); rm.close(); }