HadoopArchiveLogs implements Tool { @VisibleForTesting void filterAppsByAggregatedStatus() throws IOException, YarnException { YarnClient client = YarnClient.createYarnClient(); try { client.init(getConf()); client.start(); for (Iterator<AppInfo> it = eligibleApplications.iterator(); it.hasNext();) { AppInfo app = it.next(); try { ApplicationReport report = client.getApplicationReport( ConverterUtils.toApplicationId(app.getAppId())); LogAggregationStatus aggStatus = report.getLogAggregationStatus(); if (aggStatus.equals(LogAggregationStatus.RUNNING) || aggStatus.equals(LogAggregationStatus.RUNNING_WITH_FAILURE) || aggStatus.equals(LogAggregationStatus.NOT_START) || aggStatus.equals(LogAggregationStatus.DISABLED) || aggStatus.equals(LogAggregationStatus.FAILED)) { if (verbose) { LOG.info("Skipping " + app.getAppId() + " due to aggregation status being " + aggStatus); } it.remove(); } else { if (verbose) { LOG.info(app.getAppId() + " has aggregation status " + aggStatus); } app.setFinishTime(report.getFinishTime()); } } catch (ApplicationNotFoundException e) { if (verbose) { LOG.info(app.getAppId() + " not in the ResourceManager"); } } } } finally { if (client != null) { client.stop(); } } } HadoopArchiveLogs(Configuration conf); static void main(String[] args); @Override int run(String[] args); @Override void setConf(Configuration conf); @Override Configuration getConf();  }
@Test(timeout = 30000) public void testFilterAppsByAggregatedStatus() throws Exception { MiniYARNCluster yarnCluster = null; try { Configuration conf = new Configuration(); conf.setBoolean(YarnConfiguration.LOG_AGGREGATION_ENABLED, true); yarnCluster = new MiniYARNCluster(TestHadoopArchiveLogs.class.getSimpleName(), 1, 1, 1, 1); yarnCluster.init(conf); yarnCluster.start(); conf = yarnCluster.getConfig(); RMContext rmContext = yarnCluster.getResourceManager().getRMContext(); RMAppImpl appImpl1 = (RMAppImpl)createRMApp(1, conf, rmContext, LogAggregationStatus.DISABLED); RMAppImpl appImpl2 = (RMAppImpl)createRMApp(2, conf, rmContext, LogAggregationStatus.FAILED); RMAppImpl appImpl3 = (RMAppImpl)createRMApp(3, conf, rmContext, LogAggregationStatus.NOT_START); RMAppImpl appImpl4 = (RMAppImpl)createRMApp(4, conf, rmContext, LogAggregationStatus.SUCCEEDED); RMAppImpl appImpl5 = (RMAppImpl)createRMApp(5, conf, rmContext, LogAggregationStatus.RUNNING); RMAppImpl appImpl6 = (RMAppImpl)createRMApp(6, conf, rmContext, LogAggregationStatus.RUNNING_WITH_FAILURE); RMAppImpl appImpl7 = (RMAppImpl)createRMApp(7, conf, rmContext, LogAggregationStatus.TIME_OUT); RMAppImpl appImpl8 = (RMAppImpl)createRMApp(8, conf, rmContext, LogAggregationStatus.SUCCEEDED); rmContext.getRMApps().put(appImpl1.getApplicationId(), appImpl1); rmContext.getRMApps().put(appImpl2.getApplicationId(), appImpl2); rmContext.getRMApps().put(appImpl3.getApplicationId(), appImpl3); rmContext.getRMApps().put(appImpl4.getApplicationId(), appImpl4); rmContext.getRMApps().put(appImpl5.getApplicationId(), appImpl5); rmContext.getRMApps().put(appImpl6.getApplicationId(), appImpl6); rmContext.getRMApps().put(appImpl7.getApplicationId(), appImpl7); HadoopArchiveLogs hal = new HadoopArchiveLogs(conf); Assert.assertEquals(0, hal.eligibleApplications.size()); hal.eligibleApplications.add( new HadoopArchiveLogs.AppInfo(appImpl1.getApplicationId().toString(), USER)); hal.eligibleApplications.add( new HadoopArchiveLogs.AppInfo(appImpl2.getApplicationId().toString(), USER)); hal.eligibleApplications.add( new HadoopArchiveLogs.AppInfo(appImpl3.getApplicationId().toString(), USER)); HadoopArchiveLogs.AppInfo app4 = new HadoopArchiveLogs.AppInfo(appImpl4.getApplicationId().toString(), USER); hal.eligibleApplications.add(app4); hal.eligibleApplications.add( new HadoopArchiveLogs.AppInfo(appImpl5.getApplicationId().toString(), USER)); hal.eligibleApplications.add( new HadoopArchiveLogs.AppInfo(appImpl6.getApplicationId().toString(), USER)); HadoopArchiveLogs.AppInfo app7 = new HadoopArchiveLogs.AppInfo(appImpl7.getApplicationId().toString(), USER); hal.eligibleApplications.add(app7); HadoopArchiveLogs.AppInfo app8 = new HadoopArchiveLogs.AppInfo(appImpl8.getApplicationId().toString(), USER); hal.eligibleApplications.add(app8); Assert.assertEquals(8, hal.eligibleApplications.size()); hal.filterAppsByAggregatedStatus(); Assert.assertEquals(3, hal.eligibleApplications.size()); Assert.assertTrue(hal.eligibleApplications.contains(app4)); Assert.assertTrue(hal.eligibleApplications.contains(app7)); Assert.assertTrue(hal.eligibleApplications.contains(app8)); } finally { if (yarnCluster != null) { yarnCluster.stop(); } } }