@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);
  }
@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());
  }