code: public List<AppSummary> getNewApps(JobHistoryService jhs, String cluster,
        String user, long startTime, long endTime, int limit) throws IOException {
      byte[] startRow = null;
      if (StringUtils.isNotBlank(user)) {
        startRow = ByteUtil.join(Constants.SEP_BYTES, Bytes.toBytes(cluster),
            Bytes.toBytes(user));
      } else {
        startRow = ByteUtil.join(Constants.SEP_BYTES, Bytes.toBytes(cluster));
      }
      LOG.info(
          "Reading app version rows start at " + Bytes.toStringBinary(startRow));
      Scan scan = new Scan();
      // start scanning app version table at cluster!user!
      scan.setStartRow(startRow);
      // require that all results match this flow prefix
      FilterList filters = new FilterList(FilterList.Operator.MUST_PASS_ALL);
      filters.addFilter(new WhileMatchFilter(new PrefixFilter(startRow)));
  
      scan.setFilter(filters);
  
      List<AppKey> newAppsKeys = new ArrayList<AppKey>();
      try {
        newAppsKeys =
            createNewAppKeysFromResults(scan, startTime, endTime, limit);
      } catch (IOException e) {
        LOG.error(
            "Caught exception while trying to scan, returning empty list of flows: "
                + e.toString());
      }
  
      List<AppSummary> newApps = new ArrayList<AppSummary>();
      for (AppKey ak : newAppsKeys) {
        AppSummary anApp = new AppSummary(ak);
        List<Flow> flows =
            jhs.getFlowSeries(ak.getCluster(), ak.getUserName(), ak.getAppId(),
                null, Boolean.FALSE, startTime, endTime, Integer.MAX_VALUE);
        for (Flow f : flows) {
          anApp.addFlow(f);
        }
        newApps.add(anApp);
      }
      return newApps;
  
    }
docs: scans the app version table to look for jobs that showed up in the given
  time range creates the flow key that maps to these apps
  @param cluster
  @param user
  @param startTime
  @param endTime
  @param limit
  @return list of flow keys
  @throws IOException
  @throws ProcessingException