JobHistory extends AbstractService implements HistoryContext { public void refreshJobRetentionSettings() { if (getServiceState() == STATE.STARTED) { conf = createConf(); long maxHistoryAge = conf.getLong(JHAdminConfig.MR_HISTORY_MAX_AGE_MS, JHAdminConfig.DEFAULT_MR_HISTORY_MAX_AGE); hsManager.setMaxHistoryAge(maxHistoryAge); if (futureHistoryCleaner != null) { futureHistoryCleaner.cancel(false); } futureHistoryCleaner = null; scheduleHistoryCleaner(); } else { LOG.warn("Failed to execute refreshJobRetentionSettings : Job History service is not started"); } } JobHistory(); @Override String getApplicationName(); @Override Job getJob(JobId jobId); @Override Map<JobId, Job> getAllJobs(ApplicationId appID); @Override Map<JobId, Job> getAllJobs(); void refreshLoadedJobCache(); @Override JobsInfo getPartialJobs(Long offset, Long count, String user,
      String queue, Long sBegin, Long sEnd, Long fBegin, Long fEnd,
      JobState jobState); void refreshJobRetentionSettings(); long getCleanerInterval(); @Override ApplicationAttemptId getApplicationAttemptId(); @Override ApplicationId getApplicationID(); @Override EventHandler<Event> getEventHandler(); @Override CharSequence getUser(); @Override Clock getClock(); @Override ClusterInfo getClusterInfo(); @Override Set<String> getBlacklistedNodes(); @Override ClientToAMTokenSecretManager getClientToAMTokenSecretManager(); @Override boolean isLastAMRetry(); @Override boolean hasSuccessfullyUnregistered(); @Override String getNMHostname(); @Override TaskAttemptFinishingMonitor getTaskAttemptFinishingMonitor(); @Override String getHistoryUrl(); @Override void setHistoryUrl(String historyUrl); static final Pattern CONF_FILENAME_REGEX; static final String OLD_SUFFIX; }
@Test public void testRefreshJobRetentionSettings() throws IOException, InterruptedException { String root = "mockfs: String historyDoneDir = root + "mapred/history/done"; long now = System.currentTimeMillis(); long someTimeYesterday = now - (25l * 3600 * 1000); long timeBefore200Secs = now - (200l * 1000); String timestampComponent = JobHistoryUtils .timestampDirectoryComponent(someTimeYesterday); Path donePathYesterday = new Path(historyDoneDir, timestampComponent + "/" + "000000"); FileStatus dirCreatedYesterdayStatus = new FileStatus(0, true, 0, 0, someTimeYesterday, donePathYesterday); timestampComponent = JobHistoryUtils .timestampDirectoryComponent(timeBefore200Secs); Path donePathToday = new Path(historyDoneDir, timestampComponent + "/" + "000000"); FileStatus dirCreatedTodayStatus = new FileStatus(0, true, 0, 0, timeBefore200Secs, donePathToday); Path fileUnderYesterdayDir = new Path(donePathYesterday.toString(), "job_1372363578825_0015-" + someTimeYesterday + "-user-Sleep+job-" + someTimeYesterday + "-1-1-SUCCEEDED-default.jhist"); FileStatus fileUnderYesterdayDirStatus = new FileStatus(10, false, 0, 0, someTimeYesterday, fileUnderYesterdayDir); Path fileUnderTodayDir = new Path(donePathYesterday.toString(), "job_1372363578825_0016-" + timeBefore200Secs + "-user-Sleep+job-" + timeBefore200Secs + "-1-1-SUCCEEDED-default.jhist"); FileStatus fileUnderTodayDirStatus = new FileStatus(10, false, 0, 0, timeBefore200Secs, fileUnderTodayDir); HistoryFileManager historyManager = spy(new HistoryFileManager()); jobHistory = spy(new JobHistory()); List<FileStatus> fileStatusList = new LinkedList<FileStatus>(); fileStatusList.add(dirCreatedYesterdayStatus); fileStatusList.add(dirCreatedTodayStatus); doReturn(4).when(jobHistory).getInitDelaySecs(); doReturn(historyManager).when(jobHistory).createHistoryFileManager(); List<FileStatus> list1 = new LinkedList<FileStatus>(); list1.add(fileUnderYesterdayDirStatus); doReturn(list1).when(historyManager).scanDirectoryForHistoryFiles( eq(donePathYesterday), any(FileContext.class)); List<FileStatus> list2 = new LinkedList<FileStatus>(); list2.add(fileUnderTodayDirStatus); doReturn(list2).when(historyManager).scanDirectoryForHistoryFiles( eq(donePathToday), any(FileContext.class)); doReturn(fileStatusList).when(historyManager) .getHistoryDirsForCleaning(Mockito.anyLong()); doReturn(true).when(historyManager).deleteDir(any(FileStatus.class)); JobListCache jobListCache = mock(JobListCache.class); HistoryFileInfo fileInfo = mock(HistoryFileInfo.class); doReturn(jobListCache).when(historyManager).createJobListCache(); when(jobListCache.get(any(JobId.class))).thenReturn(fileInfo); doNothing().when(fileInfo).delete(); Configuration conf = new Configuration(); conf.setLong(JHAdminConfig.MR_HISTORY_MAX_AGE_MS, 24l * 3600 * 1000); conf.setLong(JHAdminConfig.MR_HISTORY_CLEANER_INTERVAL_MS, 2 * 1000); jobHistory.init(conf); jobHistory.start(); assertEquals(2 * 1000l, jobHistory.getCleanerInterval()); verify(fileInfo, timeout(20000).times(1)).delete(); fileStatusList.remove(dirCreatedYesterdayStatus); conf.setLong(JHAdminConfig.MR_HISTORY_MAX_AGE_MS, 10 * 1000); conf.setLong(JHAdminConfig.MR_HISTORY_CLEANER_INTERVAL_MS, 1 * 1000); doReturn(conf).when(jobHistory).createConf(); jobHistory.refreshJobRetentionSettings(); assertEquals(1 * 1000l, jobHistory.getCleanerInterval()); verify(fileInfo, timeout(20000).times(2)).delete(); }