@Private
  public void handleEvent(JobHistoryEvent event) {
    synchronized (lock) {

      // If this is JobSubmitted Event, setup the writer
      if (event.getHistoryEvent().getEventType() == EventType.AM_STARTED) {
        try {
          AMStartedEvent amStartedEvent =
              (AMStartedEvent) event.getHistoryEvent();
          setupEventWriter(event.getJobID(), amStartedEvent);
        } catch (IOException ioe) {
          LOG.error("Error JobHistoryEventHandler in handleEvent: " + event,
              ioe);
          throw new YarnRuntimeException(ioe);
        }
      }

      // For all events
      // (1) Write it out
      // (2) Process it for JobSummary
      // (3) Process it for ATS (if enabled)
      MetaInfo mi = fileMap.get(event.getJobID());
      try {
        HistoryEvent historyEvent = event.getHistoryEvent();
        if (! (historyEvent instanceof NormalizedResourceEvent)) {
          mi.writeEvent(historyEvent);
        }
        processEventForJobSummary(event.getHistoryEvent(), mi.getJobSummary(),
            event.getJobID());
        if (LOG.isDebugEnabled()) {
          LOG.debug("In HistoryEventHandler "
              + event.getHistoryEvent().getEventType());
        }
      } catch (IOException e) {
        LOG.error("Error writing History Event: " + event.getHistoryEvent(),
            e);
        throw new YarnRuntimeException(e);
      }

      if (event.getHistoryEvent().getEventType() == EventType.JOB_SUBMITTED) {
        JobSubmittedEvent jobSubmittedEvent =
            (JobSubmittedEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setSubmitTime(jobSubmittedEvent.getSubmitTime());
        mi.getJobIndexInfo().setQueueName(jobSubmittedEvent.getJobQueueName());
      }
      //initialize the launchTime in the JobIndexInfo of MetaInfo
      if(event.getHistoryEvent().getEventType() == EventType.JOB_INITED ){
        JobInitedEvent jie = (JobInitedEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setJobStartTime(jie.getLaunchTime());
      }
      
      if (event.getHistoryEvent().getEventType() == EventType.JOB_QUEUE_CHANGED) {
        JobQueueChangeEvent jQueueEvent =
            (JobQueueChangeEvent) event.getHistoryEvent();
        mi.getJobIndexInfo().setQueueName(jQueueEvent.getJobQueueName());
      }

      // If this is JobFinishedEvent, close the writer and setup the job-index
      if (event.getHistoryEvent().getEventType() == EventType.JOB_FINISHED) {
        try {
          JobFinishedEvent jFinishedEvent =
              (JobFinishedEvent) event.getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jFinishedEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jFinishedEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(
              jFinishedEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(JobState.SUCCEEDED.toString());
          closeEventWriter(event.getJobID());
          processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }
      // In case of JOB_ERROR, only process all the Done files(e.g. job
      // summary, job history file etc.) if it is last AM retry.
      if (event.getHistoryEvent().getEventType() == EventType.JOB_ERROR) {
        try {
          JobUnsuccessfulCompletionEvent jucEvent =
              (JobUnsuccessfulCompletionEvent) event.getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
          closeEventWriter(event.getJobID());
          if(context.isLastAMRetry())
            processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }

      if (event.getHistoryEvent().getEventType() == EventType.JOB_FAILED
          || event.getHistoryEvent().getEventType() == EventType.JOB_KILLED) {
        try {
          JobUnsuccessfulCompletionEvent jucEvent = 
              (JobUnsuccessfulCompletionEvent) event
              .getHistoryEvent();
          mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
          mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
          closeEventWriter(event.getJobID());
          processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }
    }
  }
@Test
  public void testPropertyRedactionForJHS() throws Exception {
    final Configuration conf = new Configuration();

    String sensitivePropertyName = "aws.fake.credentials.name";
    String sensitivePropertyValue = "aws.fake.credentials.val";
    conf.set(sensitivePropertyName, sensitivePropertyValue);
    conf.set(MRJobConfig.MR_JOB_REDACTED_PROPERTIES,
        sensitivePropertyName);
    conf.set(CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY,
        dfsCluster.getURI().toString());
    final TestParams params = new TestParams();
    conf.set(MRJobConfig.MR_AM_STAGING_DIR, params.dfsWorkDir);

    final JHEvenHandlerForTest jheh =
        new JHEvenHandlerForTest(params.mockAppContext, 0, false);

    try {
      jheh.init(conf);
      jheh.start();
      handleEvent(jheh, new JobHistoryEvent(params.jobId,
          new AMStartedEvent(params.appAttemptId, 200, params.containerId,
              "nmhost", 3000, 4000, -1)));
      handleEvent(jheh, new JobHistoryEvent(params.jobId,
          new JobUnsuccessfulCompletionEvent(TypeConverter.fromYarn(
              params.jobId), 0, 0, 0, 0, 0, 0, 0,
              JobStateInternal.FAILED.toString())));

      // verify the value of the sensitive property in job.xml is restored.
      assertThat(conf.get(sensitivePropertyName))
          .isEqualTo(sensitivePropertyValue)
          .withFailMessage(sensitivePropertyName + " is modified.");

      // load the job_conf.xml in JHS directory and verify property redaction.
      Path jhsJobConfFile = getJobConfInIntermediateDoneDir(conf, params.jobId);
      Assert.assertTrue("The job_conf.xml file is not in the JHS directory",
          FileContext.getFileContext(conf).util().exists(jhsJobConfFile));
      Configuration jhsJobConf = new Configuration();

      try (InputStream input = FileSystem.get(conf).open(jhsJobConfFile)) {
        jhsJobConf.addResource(input);
        Assert.assertEquals(
            sensitivePropertyName + " is not redacted in HDFS.",
            MRJobConfUtil.REDACTION_REPLACEMENT_VAL,
            jhsJobConf.get(sensitivePropertyName));
      }
    } finally {
      jheh.stop();
      purgeHdfsHistoryIntermediateDoneDirectory(conf);
    }
  }