@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.addToCluserNodeLabels(ImmutableSet.of("x", "y")); 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); MockNM lostNode = rm.registerNode("host2:1235", 1024); rm.sendNodeStarted(lostNode); lostNode.nodeHeartbeat(true); rm.NMwaitForState(lostNode.getNodeId(), NodeState.RUNNING); rm.sendNodeLost(lostNode); 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); 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()); Assert.assertTrue(nodeReports.get(0).getNodeLabels().contains("x")); node.nodeHeartbeat(false); nodeReports = client.getClusterNodes(request).getNodeReports(); Assert.assertEquals("Unhealthy nodes should not show up by default", 0, nodeReports.size()); map = new HashMap<NodeId, Set<String>>(); map.put(node.getNodeId(), ImmutableSet.of("y")); labelsMgr.replaceLabelsOnNode(map); 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")); map = new HashMap<NodeId, Set<String>>(); map.put(node.getNodeId(), ImmutableSet.of("y")); labelsMgr.removeLabelsFromNode(map); rm.registerNode("host3:1236", 1024); request = GetClusterNodesRequest.newInstance(EnumSet.allOf(NodeState.class)); nodeReports = client.getClusterNodes(request).getNodeReports(); Assert.assertEquals(3, nodeReports.size()); for (NodeReport report : nodeReports) { Assert.assertTrue(report.getNodeLabels() != null && report.getNodeLabels().isEmpty()); } rpc.stopProxy(client, conf); rm.close(); }
@Override public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request) throws YarnException { GetClusterNodesResponse response = recordFactory.newRecordInstance(GetClusterNodesResponse.class); EnumSet<NodeState> nodeStates = request.getNodeStates(); if (nodeStates == null || nodeStates.isEmpty()) { nodeStates = EnumSet.allOf(NodeState.class); } Collection<RMNode> nodes = RMServerUtils.queryRMNodes(rmContext, nodeStates); List<NodeReport> nodeReports = new ArrayList<NodeReport>(nodes.size()); for (RMNode nodeInfo : nodes) { nodeReports.add(createNodeReports(nodeInfo)); } response.setNodeReports(nodeReports); return response; }
ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request) throws YarnException { GetClusterNodesResponse response = recordFactory.newRecordInstance(GetClusterNodesResponse.class); EnumSet<NodeState> nodeStates = request.getNodeStates(); if (nodeStates == null || nodeStates.isEmpty()) { nodeStates = EnumSet.allOf(NodeState.class); } Collection<RMNode> nodes = RMServerUtils.queryRMNodes(rmContext, nodeStates); List<NodeReport> nodeReports = new ArrayList<NodeReport>(nodes.size()); for (RMNode nodeInfo : nodes) { nodeReports.add(createNodeReports(nodeInfo)); } response.setNodeReports(nodeReports); return response; } }
ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request) throws YarnException { GetClusterNodesResponse response = recordFactory.newRecordInstance(GetClusterNodesResponse.class); EnumSet<NodeState> nodeStates = request.getNodeStates(); if (nodeStates == null || nodeStates.isEmpty()) { nodeStates = EnumSet.allOf(NodeState.class); } Collection<RMNode> nodes = RMServerUtils.queryRMNodes(rmContext, nodeStates); List<NodeReport> nodeReports = new ArrayList<NodeReport>(nodes.size()); for (RMNode nodeInfo : nodes) { nodeReports.add(createNodeReports(nodeInfo)); } response.setNodeReports(nodeReports); 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); }
ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request) throws YarnException { GetClusterNodesResponse response = recordFactory.newRecordInstance(GetClusterNodesResponse.class); EnumSet<NodeState> nodeStates = request.getNodeStates(); if (nodeStates == null || nodeStates.isEmpty()) { nodeStates = EnumSet.allOf(NodeState.class); } Collection<RMNode> nodes = RMServerUtils.queryRMNodes(rmContext, nodeStates); List<NodeReport> nodeReports = new ArrayList<NodeReport>(nodes.size()); for (RMNode nodeInfo : nodes) { nodeReports.add(createNodeReports(nodeInfo)); } response.setNodeReports(nodeReports); 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 KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetApplicationsResponse getApplications(
      GetApplicationsRequest request); @Private GetApplicationsResponse getApplications(
      GetApplicationsRequest request, boolean caseSensitive); @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 ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request); }
ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetClusterNodesResponse getClusterNodes(GetClusterNodesRequest request) throws YarnException { GetClusterNodesResponse response = recordFactory.newRecordInstance(GetClusterNodesResponse.class); EnumSet<NodeState> nodeStates = request.getNodeStates(); if (nodeStates == null || nodeStates.isEmpty()) { nodeStates = EnumSet.allOf(NodeState.class); } Collection<RMNode> nodes = RMServerUtils.queryRMNodes(rmContext, nodeStates); List<NodeReport> nodeReports = new ArrayList<NodeReport>(nodes.size()); for (RMNode nodeInfo : nodes) { nodeReports.add(createNodeReports(nodeInfo)); } response.setNodeReports(nodeReports); 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 KillApplicationResponse forceKillApplication(
      KillApplicationRequest request); @Override GetClusterMetricsResponse getClusterMetrics(
      GetClusterMetricsRequest request); @Override GetApplicationsResponse getApplications(
      GetApplicationsRequest request); @Private GetApplicationsResponse getApplications(
      GetApplicationsRequest request, boolean caseSensitive); @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 ReservationSubmissionResponse submitReservation(
      ReservationSubmissionRequest request); @Override ReservationUpdateResponse updateReservation(
      ReservationUpdateRequest request); @Override ReservationDeleteResponse deleteReservation(
      ReservationDeleteRequest request); @Override GetNodesToLabelsResponse getNodeToLabels(
      GetNodesToLabelsRequest request); @Override GetClusterNodeLabelsResponse getClusterNodeLabels(
      GetClusterNodeLabelsRequest request);  }