RMServerUtils { public static List<RMNode> queryRMNodes(RMContext context, EnumSet<NodeState> acceptedStates) { ArrayList<RMNode> results = new ArrayList<RMNode>(); boolean hasActive = false; boolean hasInactive = false; for (NodeState nodeState : acceptedStates) { if (!hasInactive && nodeState.isInactiveState()) { hasInactive = true; } if (!hasActive && nodeState.isActiveState()) { hasActive = true; } if (hasActive && hasInactive) { break; } } if (hasActive) { for (RMNode rmNode : context.getRMNodes().values()) { if (acceptedStates.contains(rmNode.getState())) { results.add(rmNode); } } } if (hasInactive) { for (RMNode rmNode : context.getInactiveRMNodes().values()) { if ((rmNode != null) && acceptedStates.contains(rmNode.getState())) { results.add(rmNode); } } } return results; }  static List<RMNode> queryRMNodes(RMContext context,
      EnumSet<NodeState> acceptedStates); static ContainerUpdates validateAndSplitUpdateResourceRequests(RMContext rmContext,
      AllocateRequest request, Resource maximumAllocation,
      List<UpdateContainerError> updateErrors); static void normalizeAndValidateRequests(List<ResourceRequest> ask,
      Resource maximumAllocation, String queueName, YarnScheduler scheduler,
      RMContext rmContext, boolean nodeLabelsEnabled); static void checkSchedContainerChangeRequest(
      SchedContainerChangeRequest request, boolean increase); static void validateBlacklistRequest(
      ResourceBlacklistRequest blacklistRequest); static void validateContainerReleaseRequest(List<ContainerId> containerReleaseList,
      ApplicationAttemptId appAttemptId); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, final Logger LOG); static UserGroupInformation verifyAdminAccess(
      YarnAuthorizationProvider authorizer, String method, String module,
      final Logger LOG); static YarnApplicationState createApplicationState(
      RMAppState rmAppState); static YarnApplicationAttemptState createApplicationAttemptState(
      RMAppAttemptState rmAppAttemptState); static void processRMProxyUsersConf(Configuration conf); static void validateApplicationTimeouts(
      Map<ApplicationTimeoutType, Long> timeouts); static Map<ApplicationTimeoutType, Long> validateISO8601AndConvertToLocalTimeEpoch(
      Map<ApplicationTimeoutType, String> timeoutsInISO8601); static int getApplicableNodeCountForAM(RMContext rmContext,
      Configuration conf, List<ResourceRequest> amReqs); static Long getOrDefault(Map<String, Long> map, String key,
      Long defaultValue); static final String UPDATE_OUTSTANDING_ERROR; static final String RESOURCE_OUTSIDE_ALLOWED_RANGE; static final ApplicationResourceUsageReport DUMMY_APPLICATION_RESOURCE_USAGE_REPORT; }
@Test public void testQueryRMNodes() throws Exception { RMContext rmContext = mock(RMContext.class); NodeId node1 = NodeId.newInstance("node1", 1234); RMNode rmNode1 = mock(RMNode.class); ConcurrentMap<NodeId, RMNode> inactiveList = new ConcurrentHashMap<NodeId, RMNode>(); when(rmNode1.getState()).thenReturn(NodeState.SHUTDOWN); inactiveList.put(node1, rmNode1); when(rmContext.getInactiveRMNodes()).thenReturn(inactiveList); List<RMNode> result = RMServerUtils.queryRMNodes(rmContext, EnumSet.of(NodeState.SHUTDOWN)); Assert.assertTrue(result.size() != 0); assertThat(result.get(0)).isEqualTo(rmNode1); when(rmNode1.getState()).thenReturn(NodeState.DECOMMISSIONED); result = RMServerUtils.queryRMNodes(rmContext, EnumSet.of(NodeState.DECOMMISSIONED)); Assert.assertTrue(result.size() != 0); assertThat(result.get(0)).isEqualTo(rmNode1); when(rmNode1.getState()).thenReturn(NodeState.LOST); result = RMServerUtils.queryRMNodes(rmContext, EnumSet.of(NodeState.LOST)); Assert.assertTrue(result.size() != 0); assertThat(result.get(0)).isEqualTo(rmNode1); when(rmNode1.getState()).thenReturn(NodeState.REBOOTED); result = RMServerUtils.queryRMNodes(rmContext, EnumSet.of(NodeState.REBOOTED)); Assert.assertTrue(result.size() != 0); assertThat(result.get(0)).isEqualTo(rmNode1); }