AggregatedLogDeletionService extends AbstractService { public void refreshLogRetentionSettings() throws IOException { if (getServiceState() == STATE.STARTED) { Configuration conf = createConf(); setConfig(conf); stopRMClient(); stopTimer(); scheduleLogDeletionTask(); } else { LOG.warn("Failed to execute refreshLogRetentionSettings : Aggregated Log Deletion Service is not started"); } } AggregatedLogDeletionService(); void refreshLogRetentionSettings(); long getCheckIntervalMsecs();  }
@Test public void testRefreshLogRetentionSettings() throws Exception { long now = System.currentTimeMillis(); long before2000Secs = now - (2000 * 1000); long before50Secs = now - (50 * 1000); String root = "mockfs: String remoteRootLogDir = root + "tmp/logs"; String suffix = "logs"; String newSuffix = LogAggregationUtils.getBucketSuffix() + suffix; final Configuration conf = new Configuration(); conf.setClass("fs.mockfs.impl", MockFileSystem.class, FileSystem.class); conf.set(YarnConfiguration.LOG_AGGREGATION_ENABLED, "true"); conf.set(YarnConfiguration.LOG_AGGREGATION_RETAIN_SECONDS, "1800"); conf.set(YarnConfiguration.LOG_AGGREGATION_RETAIN_CHECK_INTERVAL_SECONDS, "1"); conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR, remoteRootLogDir); conf.set(YarnConfiguration.NM_REMOTE_APP_LOG_DIR_SUFFIX, suffix); conf.set(YarnConfiguration.LOG_AGGREGATION_FILE_FORMATS, "TFile"); conf.set(String.format(LOG_AGGREGATION_FILE_CONTROLLER_FMT, "TFile"), LogAggregationTFileController.class.getName()); Path rootPath = new Path(root); FileSystem rootFs = rootPath.getFileSystem(conf); FileSystem mockFs = ((FilterFileSystem) rootFs).getRawFileSystem(); Path remoteRootLogPath = new Path(remoteRootLogDir); Path userDir = new Path(remoteRootLogPath, "me"); FileStatus userDirStatus = new FileStatus(0, true, 0, 0, before50Secs, userDir); when(mockFs.listStatus(remoteRootLogPath)).thenReturn( new FileStatus[] { userDirStatus }); Path suffixDir = new Path(userDir, newSuffix); FileStatus suffixStatus = new FileStatus(0, true, 0, 0, before50Secs, suffixDir); ApplicationId appId1 = ApplicationId.newInstance(System.currentTimeMillis(), 1); Path app1Dir = LogAggregationUtils.getRemoteAppLogDir( remoteRootLogPath, appId1, "me", suffix); Path bucketDir = LogAggregationUtils.getRemoteBucketDir( remoteRootLogPath, "me", suffix, appId1); FileStatus bucketDirStatus = new FileStatus(0, true, 0, 0, before50Secs, bucketDir); FileStatus app1DirStatus = new FileStatus(0, true, 0, 0, before2000Secs, app1Dir); ApplicationId appId2 = ApplicationId.newInstance(System.currentTimeMillis(), 2); Path app2Dir = LogAggregationUtils.getRemoteAppLogDir( remoteRootLogPath, appId2, "me", suffix); FileStatus app2DirStatus = new FileStatus(0, true, 0, 0, before50Secs, app2Dir); when(mockFs.listStatus(userDir)).thenReturn( new FileStatus[] {suffixStatus }); when(mockFs.listStatus(suffixDir)).thenReturn( new FileStatus[] {bucketDirStatus }); when(mockFs.listStatus(bucketDir)).thenReturn( new FileStatus[] {app1DirStatus, app2DirStatus }); Path app1Log1 = new Path(app1Dir, "host1"); FileStatus app1Log1Status = new FileStatus(10, false, 1, 1, before2000Secs, app1Log1); when(mockFs.listStatus(app1Dir)).thenReturn( new FileStatus[] { app1Log1Status }); Path app2Log1 = new Path(app2Dir, "host1"); FileStatus app2Log1Status = new FileStatus(10, false, 1, 1, before50Secs, app2Log1); when(mockFs.listStatus(app2Dir)).thenReturn( new FileStatus[] { app2Log1Status }); final List<ApplicationId> finishedApplications = Collections.unmodifiableList(Arrays.asList(appId1, appId2)); AggregatedLogDeletionService deletionSvc = new AggregatedLogDeletionService() { @Override protected Configuration createConf() { return conf; } @Override protected ApplicationClientProtocol createRMClient() throws IOException { try { return createMockRMClient(finishedApplications, null); } catch (Exception e) { throw new IOException(e); } } @Override protected void stopRMClient() { } }; deletionSvc.init(conf); deletionSvc.start(); verify(mockFs, timeout(10000)).delete(app1Dir, true); verify(mockFs, timeout(3000).times(0)).delete(app2Dir, true); conf.set(YarnConfiguration.LOG_AGGREGATION_RETAIN_SECONDS, "50"); conf.set(YarnConfiguration.LOG_AGGREGATION_RETAIN_CHECK_INTERVAL_SECONDS, "2"); Assert.assertTrue(2000l != deletionSvc.getCheckIntervalMsecs()); deletionSvc.refreshLogRetentionSettings(); Assert.assertTrue(2000l == deletionSvc.getCheckIntervalMsecs()); verify(mockFs, timeout(10000)).delete(app2Dir, true); deletionSvc.stop(); }