ClientRMService extends AbstractService implements
    ApplicationClientProtocol { @Override public GetApplicationsResponse getApplications(GetApplicationsRequest request) throws YarnException { UserGroupInformation callerUGI; try { callerUGI = UserGroupInformation.getCurrentUser(); } catch (IOException ie) { LOG.info("Error getting UGI ", ie); throw RPCUtil.getRemoteException(ie); } Set<String> applicationTypes = getLowerCasedAppTypes(request); EnumSet<YarnApplicationState> applicationStates = request.getApplicationStates(); Set<String> users = request.getUsers(); Set<String> queues = request.getQueues(); Set<String> tags = request.getApplicationTags(); long limit = request.getLimit(); Range<Long> start = request.getStartRange(); Range<Long> finish = request.getFinishRange(); ApplicationsRequestScope scope = request.getScope(); final Map<ApplicationId, RMApp> apps = rmContext.getRMApps(); Iterator<RMApp> appsIter; if (queues != null && !queues.isEmpty()) { final List<List<ApplicationAttemptId>> queueAppLists = new ArrayList<List<ApplicationAttemptId>>(); for (String queue : queues) { List<ApplicationAttemptId> appsInQueue = scheduler.getAppsInQueue(queue); if (appsInQueue != null && !appsInQueue.isEmpty()) { queueAppLists.add(appsInQueue); } } appsIter = new Iterator<RMApp>() { Iterator<List<ApplicationAttemptId>> appListIter = queueAppLists.iterator(); Iterator<ApplicationAttemptId> schedAppsIter; @Override public boolean hasNext() { return (schedAppsIter != null && schedAppsIter.hasNext()) || appListIter.hasNext(); } @Override public RMApp next() { if (schedAppsIter == null || !schedAppsIter.hasNext()) { schedAppsIter = appListIter.next().iterator(); } return apps.get(schedAppsIter.next().getApplicationId()); } @Override public void remove() { throw new UnsupportedOperationException("Remove not supported"); } }; } else { appsIter = apps.values().iterator(); } List<ApplicationReport> reports = new ArrayList<ApplicationReport>(); while (appsIter.hasNext() && reports.size() < limit) { RMApp application = appsIter.next(); if (scope == ApplicationsRequestScope.OWN && !callerUGI.getUserName().equals(application.getUser())) { continue; } if (applicationTypes != null && !applicationTypes.isEmpty()) { String appTypeToMatch = StringUtils.toLowerCase(application.getApplicationType()); if (!applicationTypes.contains(appTypeToMatch)) { continue; } } if (applicationStates != null && !applicationStates.isEmpty()) { if (!applicationStates.contains(application .createApplicationState())) { continue; } } if (users != null && !users.isEmpty() && !users.contains(application.getUser())) { continue; } if (start != null && !start.contains(application.getStartTime())) { continue; } if (finish != null && !finish.contains(application.getFinishTime())) { continue; } if (tags != null && !tags.isEmpty()) { Set<String> appTags = application.getApplicationTags(); if (appTags == null || appTags.isEmpty()) { continue; } boolean match = false; for (String tag : tags) { if (appTags.contains(tag)) { match = true; break; } } if (!match) { continue; } } boolean allowAccess = checkAccess(callerUGI, application.getUser(), ApplicationAccessType.VIEW_APP, application); if (scope == ApplicationsRequestScope.VIEWABLE && !allowAccess) { continue; } if (filterAppsByUser && !allowAccess) { continue; } reports.add(application.createAndGetApplicationReport( callerUGI.getUserName(), allowAccess)); } RMAuditLogger.logSuccess(callerUGI.getUserName(), AuditConstants.GET_APPLICATIONS_REQUEST, "ClientRMService"); GetApplicationsResponse response = recordFactory.newRecordInstance(GetApplicationsResponse.class); response.setApplicationList(reports); 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 public void testGetApplications() throws Exception { ResourceScheduler scheduler = mockResourceScheduler(); RMContext rmContext = mock(RMContext.class); mockRMContext(scheduler, rmContext); RMStateStore stateStore = mock(RMStateStore.class); when(rmContext.getStateStore()).thenReturn(stateStore); doReturn(mock(RMTimelineCollectorManager.class)).when(rmContext) .getRMTimelineCollectorManager(); RMAppManager appManager = new RMAppManager(rmContext, scheduler, null, mock(ApplicationACLsManager.class), new Configuration()); when(rmContext.getDispatcher().getEventHandler()).thenReturn( new EventHandler<Event>() { public void handle(Event event) {} }); ApplicationACLsManager mockAclsManager = mock(ApplicationACLsManager.class); QueueACLsManager mockQueueACLsManager = mock(QueueACLsManager.class); when(mockQueueACLsManager.checkAccess(any(UserGroupInformation.class), any(QueueACL.class), any(RMApp.class), any(String.class), any())) .thenReturn(true); ClientRMService rmService = new ClientRMService(rmContext, scheduler, appManager, mockAclsManager, mockQueueACLsManager, null); rmService.init(new Configuration()); String[] queues = {QUEUE_1, QUEUE_2}; String[] appNames = {MockApps.newAppName(), MockApps.newAppName(), MockApps.newAppName()}; ApplicationId[] appIds = {getApplicationId(101), getApplicationId(102), getApplicationId(103)}; List<String> tags = Arrays.asList("Tag1", "Tag2", "Tag3"); long[] submitTimeMillis = new long[3]; for (int i = 0; i < appIds.length; i++) { ApplicationId appId = appIds[i]; when(mockAclsManager.checkAccess(UserGroupInformation.getCurrentUser(), ApplicationAccessType.VIEW_APP, null, appId)).thenReturn(true); SubmitApplicationRequest submitRequest = mockSubmitAppRequest( appId, appNames[i], queues[i % queues.length], new HashSet<String>(tags.subList(0, i + 1))); Thread.sleep(1); rmService.submitApplication(submitRequest); submitTimeMillis[i] = rmService.getApplicationReport( GetApplicationReportRequest.newInstance(appId)) .getApplicationReport().getStartTime(); } GetApplicationsRequest request = GetApplicationsRequest.newInstance(); assertEquals("Incorrect total number of apps", 6, rmService.getApplications(request).getApplicationList().size()); request.setLimit(1L); assertEquals("Failed to limit applications", 1, rmService.getApplications(request).getApplicationList().size()); request = GetApplicationsRequest.newInstance(); request.setStartRange(submitTimeMillis[0] + 1, System.currentTimeMillis()); assertEquals("Incorrect number of matching start range", 2, rmService.getApplications(request).getApplicationList().size()); request.setStartRange(submitTimeMillis[1] + 1, System.currentTimeMillis()); assertEquals("Incorrect number of matching start range", 1, rmService.getApplications(request).getApplicationList().size()); request.setStartRange(submitTimeMillis[2] + 1, System.currentTimeMillis()); assertEquals("Incorrect number of matching start range", 0, rmService.getApplications(request).getApplicationList().size()); request = GetApplicationsRequest.newInstance(); Set<String> queueSet = new HashSet<String>(); request.setQueues(queueSet); queueSet.add(queues[0]); assertEquals("Incorrect number of applications in queue", 2, rmService.getApplications(request).getApplicationList().size()); assertEquals("Incorrect number of applications in queue", 2, rmService.getApplications(request).getApplicationList().size()); queueSet.add(queues[1]); assertEquals("Incorrect number of applications in queue", 3, rmService.getApplications(request).getApplicationList().size()); request = GetApplicationsRequest.newInstance(); Set<String> userSet = new HashSet<String>(); request.setUsers(userSet); userSet.add("random-user-name"); assertEquals("Incorrect number of applications for user", 0, rmService.getApplications(request).getApplicationList().size()); userSet.add(UserGroupInformation.getCurrentUser().getShortUserName()); assertEquals("Incorrect number of applications for user", 3, rmService.getApplications(request).getApplicationList().size()); rmService.setDisplayPerUserApps(true); userSet.clear(); assertEquals("Incorrect number of applications for user", 6, rmService.getApplications(request).getApplicationList().size()); rmService.setDisplayPerUserApps(false); request = GetApplicationsRequest.newInstance( ApplicationsRequestScope.ALL, null, null, null, null, null, null, null, null); Set<String> tagSet = new HashSet<String>(); request.setApplicationTags(tagSet); assertEquals("Incorrect number of matching tags", 6, rmService.getApplications(request).getApplicationList().size()); tagSet = Sets.newHashSet(tags.get(0)); request.setApplicationTags(tagSet); assertEquals("Incorrect number of matching tags", 3, rmService.getApplications(request).getApplicationList().size()); tagSet = Sets.newHashSet(tags.get(1)); request.setApplicationTags(tagSet); assertEquals("Incorrect number of matching tags", 2, rmService.getApplications(request).getApplicationList().size()); tagSet = Sets.newHashSet(tags.get(2)); request.setApplicationTags(tagSet); assertEquals("Incorrect number of matching tags", 1, rmService.getApplications(request).getApplicationList().size()); request = GetApplicationsRequest.newInstance( ApplicationsRequestScope.VIEWABLE); assertEquals("Incorrect number of applications for the scope", 6, rmService.getApplications(request).getApplicationList().size()); request = GetApplicationsRequest.newInstance( ApplicationsRequestScope.OWN); assertEquals("Incorrect number of applications for the scope", 3, rmService.getApplications(request).getApplicationList().size()); }