class 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 the query filters by queues, we can avoid considering apps outside
    // of those queues by asking the scheduler for the apps in those queues.
    if (queues != null && !queues.isEmpty()) {
      // Construct an iterator over apps in given queues
      // Collect list of lists to avoid copying all apps
      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() {
          // Because queueAppLists has no empty lists, hasNext is whether the
          // current list hasNext or whether there are any remaining lists
          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();

      // Check if current application falls under the specified scope
      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;
        }
      }

      // checkAccess can grab the scheduler lock so call it last
      boolean allowAccess = checkAccess(callerUGI, application.getUser(),
          ApplicationAccessType.VIEW_APP, application);
      if (scope == ApplicationsRequestScope.VIEWABLE && !allowAccess) {
        continue;
      }

      // Given RM is configured to display apps per user, skip apps to which
      // this caller doesn't have access to view.
      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;
  }

  public  ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager);
  public  ClientRMService(RMContext rmContext, YarnScheduler scheduler,
      RMAppManager rmAppManager, ApplicationACLsManager applicationACLsManager,
      QueueACLsManager queueACLsManager,
      RMDelegationTokenSecretManager rmDTSecretManager, Clock clock);

}