@Test public void testNewJobs() throws Exception { Configuration hbaseConfiguration = UTIL.getConfiguration(); String cluster1 = "newJobsClusterName"; String appId = "getNewJobs"; String appId2 = "getNewJobs2"; String user = "testuser"; JobHistoryService jhs = new JobHistoryService(hbaseConfiguration, hbaseConnection); AppVersionService appVersionService = new AppVersionService(hbaseConnection); AppSummaryService appSummaryService = new AppSummaryService(hbaseConnection); appVersionService.addVersion(cluster1, user, appId, "v1", 10L); flowDataGen.loadFlow(cluster1, user, appId, 10L, "v1", 3, 10, idService, historyTable); appVersionService.addVersion(cluster1, user, appId, "v2", 20L); flowDataGen.loadFlow(cluster1, user, appId, 20L, "v2", 3, 10, idService, historyTable); List<AppSummary> fl = appSummaryService.getNewApps(jhs, cluster1, user, 0L, 15L, 100); assertNotNull(fl); assertEquals(1, fl.size()); appVersionService.addVersion(cluster1, user, appId, "v3", 30L); flowDataGen.loadFlow(cluster1, user, appId, 30L, "v3", 3, 10, idService, historyTable); appVersionService.addVersion(cluster1, user, appId, "v2.5", 25L); flowDataGen.loadFlow(cluster1, user, appId, 25L, "v2.5", 3, 10, idService, historyTable); fl = appSummaryService.getNewApps(jhs, cluster1, user, 20L, 35L, 100); assertNotNull(fl); assertEquals(0, fl.size()); appVersionService.addVersion(cluster1, user, appId2, "v102", 23L); flowDataGen.loadFlow(cluster1, user, appId, 23L, "v102", 3, 10, idService, historyTable); fl = appSummaryService.getNewApps(jhs, cluster1, user, 20L, 35L, 5); assertNotNull(fl); assertEquals(1, fl.size()); assertEquals(appId2, fl.get(0).getKey().getAppId()); fl = appSummaryService.getNewApps(jhs, cluster1, user, 0L, 35L, 5); assertNotNull(fl); assertEquals(2, fl.size()); for (int i = 0; i < fl.size(); i++) { String anAppId = fl.get(i).getKey().getAppId(); if (!(appId.equals(anAppId)) && !(appId2.equals(anAppId))) { throw new AssertionError("Could not find the right apps as expected"); } } }
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(); scan.setStartRow(startRow); 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; }
AppSummaryService { 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(); scan.setStartRow(startRow); 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; } }
AppSummaryService { 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(); scan.setStartRow(startRow); 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; } AppSummaryService(Connection hbaseConnection); }
AppSummaryService { 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(); scan.setStartRow(startRow); 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; } AppSummaryService(Connection hbaseConnection); List<AppSummary> getNewApps(JobHistoryService jhs, String cluster,
      String user, long startTime, long endTime, int limit); List<AppKey> createNewAppKeysFromResults(Scan scan, long startTime,
      long endTime, int maxCount); boolean aggregateJobDetails(JobDetails jobDetails,
      AggregationConstants.AGGREGATION_TYPE aggType); List<AppSummary> getAllApps(String cluster, String user,
      long startTime, long endTime, int limit); }
AppSummaryService { 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(); scan.setStartRow(startRow); 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; } AppSummaryService(Connection hbaseConnection); List<AppSummary> getNewApps(JobHistoryService jhs, String cluster,
      String user, long startTime, long endTime, int limit); List<AppKey> createNewAppKeysFromResults(Scan scan, long startTime,
      long endTime, int maxCount); boolean aggregateJobDetails(JobDetails jobDetails,
      AggregationConstants.AGGREGATION_TYPE aggType); List<AppSummary> getAllApps(String cluster, String user,
      long startTime, long endTime, int limit);  }