HistoryFileManager extends AbstractService { @Override protected void serviceInit(Configuration conf) throws Exception { this.conf = conf; int serialNumberLowDigits = 3; serialNumberFormat = ("%0" + (JobHistoryUtils.SERIAL_NUMBER_DIRECTORY_DIGITS + serialNumberLowDigits) + "d"); long maxFSWaitTime = conf.getLong( JHAdminConfig.MR_HISTORY_MAX_START_WAIT_TIME, JHAdminConfig.DEFAULT_MR_HISTORY_MAX_START_WAIT_TIME); createHistoryDirs(SystemClock.getInstance(), 10 * 1000, maxFSWaitTime); maxTasksForLoadedJob = conf.getInt( JHAdminConfig.MR_HS_LOADED_JOBS_TASKS_MAX, JHAdminConfig.DEFAULT_MR_HS_LOADED_JOBS_TASKS_MAX); this.aclsMgr = new JobACLsManager(conf); maxHistoryAge = conf.getLong(JHAdminConfig.MR_HISTORY_MAX_AGE_MS, JHAdminConfig.DEFAULT_MR_HISTORY_MAX_AGE); jobListCache = createJobListCache(); serialNumberIndex = new SerialNumberIndex(conf.getInt( JHAdminConfig.MR_HISTORY_DATESTRING_CACHE_SIZE, JHAdminConfig.DEFAULT_MR_HISTORY_DATESTRING_CACHE_SIZE)); int numMoveThreads = conf.getInt( JHAdminConfig.MR_HISTORY_MOVE_THREAD_COUNT, JHAdminConfig.DEFAULT_MR_HISTORY_MOVE_THREAD_COUNT); moveToDoneExecutor = createMoveToDoneThreadPool(numMoveThreads); super.serviceInit(conf); } HistoryFileManager(); @Override void serviceStop(); Collection<HistoryFileInfo> getAllFileInfo(); HistoryFileInfo getFileInfo(JobId jobId);  }
@Test public void testHistoryFileInfoSummaryFileNotExist() throws Exception { HistoryFileManagerTest hmTest = new HistoryFileManagerTest(); String job = "job_1410889000000_123456"; Path summaryFile = new Path(job + ".summary"); JobIndexInfo jobIndexInfo = new JobIndexInfo(); jobIndexInfo.setJobId(TypeConverter.toYarn(JobID.forName(job))); Configuration conf = dfsCluster.getConfiguration(0); conf.set(JHAdminConfig.MR_HISTORY_DONE_DIR, "/" + UUID.randomUUID()); conf.set(JHAdminConfig.MR_HISTORY_INTERMEDIATE_DONE_DIR, "/" + UUID.randomUUID()); hmTest.serviceInit(conf); HistoryFileInfo info = hmTest.getHistoryFileInfo(null, null, summaryFile, jobIndexInfo, false); info.moveToDone(); Assert.assertFalse(info.didMoveFail()); }
@Test public void testMoveToDoneAlreadyMovedSucceeds() throws Exception { HistoryFileManagerTest historyFileManager = new HistoryFileManagerTest(); long jobTimestamp = 1535436603000L; String job = "job_" + jobTimestamp + "_123456789"; String intermediateDirectory = "/" + UUID.randomUUID(); String doneDirectory = "/" + UUID.randomUUID(); Configuration conf = dfsCluster.getConfiguration(0); conf.set(JHAdminConfig.MR_HISTORY_INTERMEDIATE_DONE_DIR, intermediateDirectory); conf.set(JHAdminConfig.MR_HISTORY_DONE_DIR, doneDirectory); Path intermediateHistoryFilePath = new Path(intermediateDirectory + "/" + job + ".jhist"); Path intermediateConfFilePath = new Path(intermediateDirectory + "/" + job + "_conf.xml"); Path doneHistoryFilePath = new Path(doneDirectory + "/" + JobHistoryUtils.timestampDirectoryComponent(jobTimestamp) + "/123456/" + job + ".jhist"); Path doneConfFilePath = new Path(doneDirectory + "/" + JobHistoryUtils.timestampDirectoryComponent(jobTimestamp) + "/123456/" + job + "_conf.xml"); dfsCluster.getFileSystem().createNewFile(doneHistoryFilePath); dfsCluster.getFileSystem().createNewFile(doneConfFilePath); historyFileManager.serviceInit(conf); JobIndexInfo jobIndexInfo = new JobIndexInfo(); jobIndexInfo.setJobId(TypeConverter.toYarn(JobID.forName(job))); jobIndexInfo.setFinishTime(jobTimestamp); HistoryFileInfo info = historyFileManager.getHistoryFileInfo( intermediateHistoryFilePath, intermediateConfFilePath, null, jobIndexInfo, false); info.moveToDone(); Assert.assertFalse(info.isMovePending()); Assert.assertEquals(doneHistoryFilePath.toString(), info.getHistoryFile().toUri().getPath()); Assert.assertEquals(doneConfFilePath.toString(), info.getConfFile().toUri().getPath()); }