/**
* Licensed to the Apache Software Foundation (ASF) under one
* or more contributor license agreements.  See the NOTICE file
* distributed with this work for additional information
* regarding copyright ownership.  The ASF licenses this file
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License.  You may obtain a copy of the License at
*
*     http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package org.apache.hadoop.mapreduce.jobhistory;

import java.io.IOException;
import java.util.Collections;
import java.util.EnumSet;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;

import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.classification.InterfaceAudience.Private;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileAlreadyExistsException;
import org.apache.hadoop.fs.FileContext;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.TaskStatus;
import org.apache.hadoop.mapreduce.Counter;
import org.apache.hadoop.mapreduce.Counters;
import org.apache.hadoop.mapreduce.JobCounter;
import org.apache.hadoop.mapreduce.MRJobConfig;
import org.apache.hadoop.mapreduce.TaskType;
import org.apache.hadoop.mapreduce.TypeConverter;
import org.apache.hadoop.mapreduce.util.JobHistoryEventUtils;
import org.apache.hadoop.mapreduce.v2.api.records.JobId;
import org.apache.hadoop.mapreduce.v2.api.records.JobState;
import org.apache.hadoop.mapreduce.v2.app.AppContext;
import org.apache.hadoop.mapreduce.v2.app.MRAppMaster;
import org.apache.hadoop.mapreduce.v2.app.job.Job;
import org.apache.hadoop.mapreduce.v2.app.job.JobStateInternal;
import org.apache.hadoop.mapreduce.v2.jobhistory.FileNameIndexUtils;
import org.apache.hadoop.mapreduce.v2.jobhistory.JHAdminConfig;
import org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils;
import org.apache.hadoop.mapreduce.v2.jobhistory.JobIndexInfo;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.service.AbstractService;
import org.apache.hadoop.util.StringUtils;
import org.apache.hadoop.yarn.api.records.timeline.TimelineEntity;
import org.apache.hadoop.yarn.api.records.timeline.TimelineEvent;
import org.apache.hadoop.yarn.api.records.timeline.TimelinePutResponse;
import org.apache.hadoop.yarn.api.records.timelineservice.ApplicationEntity;
import org.apache.hadoop.yarn.api.records.timelineservice.TimelineMetric;
import org.apache.hadoop.yarn.client.api.TimelineClient;
import org.apache.hadoop.yarn.conf.YarnConfiguration;
import org.apache.hadoop.yarn.event.EventHandler;
import org.apache.hadoop.yarn.exceptions.YarnException;
import org.apache.hadoop.yarn.exceptions.YarnRuntimeException;
import org.codehaus.jackson.map.ObjectMapper;
import org.codehaus.jackson.node.JsonNodeFactory;

import com.google.common.annotations.VisibleForTesting;
import com.sun.jersey.api.client.ClientHandlerException;

/**
 * The job history events get routed to this class. This class writes the Job
 * history events to the DFS directly into a staging dir and then moved to a
 * done-dir. JobHistory implementation is in this package to access package
 * private classes.
 */
public class JobHistoryEventHandler extends AbstractService
    implements EventHandler<JobHistoryEvent> {
  private static final JsonNodeFactory FACTORY =
      new ObjectMapper().getNodeFactory();

  private final AppContext context;
  private final int startCount;

  private int eventCounter;

  // Those file systems may differ from the job configuration
  // See org.apache.hadoop.mapreduce.v2.jobhistory.JobHistoryUtils
  // #ensurePathInDefaultFileSystem
  private FileSystem stagingDirFS; // log Dir FileSystem
  private FileSystem doneDirFS; // done Dir FileSystem


  private Path stagingDirPath = null;
  private Path doneDirPrefixPath = null; // folder for completed jobs

  private int maxUnflushedCompletionEvents;
  private int postJobCompletionMultiplier;
  private long flushTimeout;
  private int minQueueSizeForBatchingFlushes; // TODO: Rename

  private int numUnflushedCompletionEvents = 0;
  private boolean isTimerActive;
  private EventWriter.WriteMode jhistMode =
      EventWriter.WriteMode.JSON;

  protected BlockingQueue<JobHistoryEvent> eventQueue =
    new LinkedBlockingQueue<JobHistoryEvent>();
  protected Thread eventHandlingThread;
  private volatile boolean stopped;
  private final Object lock = new Object();

  private static final Log LOG = LogFactory.getLog(
      JobHistoryEventHandler.class);

  protected static final Map<JobId, MetaInfo> fileMap =
    Collections.<JobId,MetaInfo>synchronizedMap(new HashMap<JobId,MetaInfo>());

  // should job completion be force when the AM shuts down?
  protected volatile boolean forceJobCompletion = false;

  protected TimelineClient timelineClient;

  private boolean timelineServiceV2Enabled = false;

  private static String MAPREDUCE_JOB_ENTITY_TYPE = "MAPREDUCE_JOB";
  private static String MAPREDUCE_TASK_ENTITY_TYPE = "MAPREDUCE_TASK";
  private static final String MAPREDUCE_TASK_ATTEMPT_ENTITY_TYPE =
      "MAPREDUCE_TASK_ATTEMPT";

  public JobHistoryEventHandler(AppContext context, int startCount) {
    super("JobHistoryEventHandler");
    this.context = context;
    this.startCount = startCount;
  }

  /* (non-Javadoc)
   * @see org.apache.hadoop.yarn.service.AbstractService#init(org.
   * apache.hadoop.conf.Configuration)
   * Initializes the FileSystem and Path objects for the log and done directories.
   * Creates these directories if they do not already exist.
   */
  @Override
  protected void serviceInit(Configuration conf) throws Exception {
    String jobId =
      TypeConverter.fromYarn(context.getApplicationID()).toString();
    
    String stagingDirStr = null;
    String doneDirStr = null;
    String userDoneDirStr = null;
    try {
      stagingDirStr = JobHistoryUtils.getConfiguredHistoryStagingDirPrefix(conf,
          jobId);
      doneDirStr =
          JobHistoryUtils.getConfiguredHistoryIntermediateDoneDirPrefix(conf);
      userDoneDirStr =
          JobHistoryUtils.getHistoryIntermediateDoneDirForUser(conf);
    } catch (IOException e) {
      LOG.error("Failed while getting the configured log directories", e);
      throw new YarnRuntimeException(e);
    }

    //Check for the existence of the history staging dir. Maybe create it. 
    try {
      stagingDirPath =
          FileContext.getFileContext(conf).makeQualified(new Path(stagingDirStr));
      stagingDirFS = FileSystem.get(stagingDirPath.toUri(), conf);
      mkdir(stagingDirFS, stagingDirPath, new FsPermission(
          JobHistoryUtils.HISTORY_STAGING_DIR_PERMISSIONS));
    } catch (IOException e) {
      LOG.error("Failed while checking for/creating  history staging path: ["
          + stagingDirPath + "]", e);
      throw new YarnRuntimeException(e);
    }

    //Check for the existence of intermediate done dir.
    Path doneDirPath = null;
    try {
      doneDirPath = FileContext.getFileContext(conf).makeQualified(new Path(doneDirStr));
      doneDirFS = FileSystem.get(doneDirPath.toUri(), conf);
      // This directory will be in a common location, or this may be a cluster
      // meant for a single user. Creating based on the conf. Should ideally be
      // created by the JobHistoryServer or as part of deployment.
      if (!doneDirFS.exists(doneDirPath)) {
      if (JobHistoryUtils.shouldCreateNonUserDirectory(conf)) {
        LOG.info("Creating intermediate history logDir: ["
            + doneDirPath
            + "] + based on conf. Should ideally be created by the JobHistoryServer: "
            + MRJobConfig.MR_AM_CREATE_JH_INTERMEDIATE_BASE_DIR);
          mkdir(
              doneDirFS,
              doneDirPath,
              new FsPermission(
            JobHistoryUtils.HISTORY_INTERMEDIATE_DONE_DIR_PERMISSIONS
                .toShort()));
          // TODO Temporary toShort till new FsPermission(FsPermissions)
          // respects
        // sticky
      } else {
          String message = "Not creating intermediate history logDir: ["
                + doneDirPath
                + "] based on conf: "
                + MRJobConfig.MR_AM_CREATE_JH_INTERMEDIATE_BASE_DIR
                + ". Either set to true or pre-create this directory with" +
                " appropriate permissions";
        LOG.error(message);
        throw new YarnRuntimeException(message);
      }
      }
    } catch (IOException e) {
      LOG.error("Failed checking for the existance of history intermediate " +
      		"done directory: [" + doneDirPath + "]");
      throw new YarnRuntimeException(e);
    }

    //Check/create user directory under intermediate done dir.
    try {
      doneDirPrefixPath =
          FileContext.getFileContext(conf).makeQualified(new Path(userDoneDirStr));
      mkdir(doneDirFS, doneDirPrefixPath, new FsPermission(
          JobHistoryUtils.HISTORY_INTERMEDIATE_USER_DIR_PERMISSIONS));
    } catch (IOException e) {
      LOG.error("Error creating user intermediate history done directory: [ "
          + doneDirPrefixPath + "]", e);
      throw new YarnRuntimeException(e);
    }

    // Maximum number of unflushed completion-events that can stay in the queue
    // before flush kicks in.
    maxUnflushedCompletionEvents =
        conf.getInt(MRJobConfig.MR_AM_HISTORY_MAX_UNFLUSHED_COMPLETE_EVENTS,
            MRJobConfig.DEFAULT_MR_AM_HISTORY_MAX_UNFLUSHED_COMPLETE_EVENTS);
    // We want to cut down flushes after job completes so as to write quicker,
    // so we increase maxUnflushedEvents post Job completion by using the
    // following multiplier.
    postJobCompletionMultiplier =
        conf.getInt(
            MRJobConfig.MR_AM_HISTORY_JOB_COMPLETE_UNFLUSHED_MULTIPLIER,
            MRJobConfig.DEFAULT_MR_AM_HISTORY_JOB_COMPLETE_UNFLUSHED_MULTIPLIER);
    // Max time until which flush doesn't take place.
    flushTimeout =
        conf.getLong(MRJobConfig.MR_AM_HISTORY_COMPLETE_EVENT_FLUSH_TIMEOUT_MS,
            MRJobConfig.DEFAULT_MR_AM_HISTORY_COMPLETE_EVENT_FLUSH_TIMEOUT_MS);
    minQueueSizeForBatchingFlushes =
        conf.getInt(
            MRJobConfig.MR_AM_HISTORY_USE_BATCHED_FLUSH_QUEUE_SIZE_THRESHOLD,
            MRJobConfig.DEFAULT_MR_AM_HISTORY_USE_BATCHED_FLUSH_QUEUE_SIZE_THRESHOLD);

    // TODO replace MR specific configurations on timeline service with getting
    // configuration from RM through registerApplicationMaster() in
    // ApplicationMasterProtocol with return value for timeline service
    // configuration status: off, on_with_v1 or on_with_v2.
    if (conf.getBoolean(MRJobConfig.MAPREDUCE_JOB_EMIT_TIMELINE_DATA,
        MRJobConfig.DEFAULT_MAPREDUCE_JOB_EMIT_TIMELINE_DATA)) {
      LOG.info("Emitting job history data to the timeline service is enabled");
      if (YarnConfiguration.timelineServiceEnabled(conf)) {

        timelineClient =
            ((MRAppMaster.RunningAppContext)context).getTimelineClient();
        timelineClient.init(conf);
        timelineServiceV2Enabled =
            YarnConfiguration.timelineServiceV2Enabled(conf);
        LOG.info("Timeline service is enabled; version: " +
            YarnConfiguration.getTimelineServiceVersion(conf));
      } else {
        LOG.info("Timeline service is not enabled");
      }
    } else {
      LOG.info("Emitting job history data to the timeline server is not " +
          "enabled");
    }

    // Flag for setting
    String jhistFormat = conf.get(JHAdminConfig.MR_HS_JHIST_FORMAT,
        JHAdminConfig.DEFAULT_MR_HS_JHIST_FORMAT);
    if (jhistFormat.equals("json")) {
      jhistMode = EventWriter.WriteMode.JSON;
    } else if (jhistFormat.equals("binary")) {
      jhistMode = EventWriter.WriteMode.BINARY;
    } else {
      LOG.warn("Unrecognized value '" + jhistFormat + "' for property " +
          JHAdminConfig.MR_HS_JHIST_FORMAT + ".  Valid values are " +
          "'json' or 'binary'.  Falling back to default value '" +
          JHAdminConfig.DEFAULT_MR_HS_JHIST_FORMAT + "'.");
    }

    super.serviceInit(conf);
  }

  private void mkdir(FileSystem fs, Path path, FsPermission fsp)
      throws IOException {
    if (!fs.exists(path)) {
      try {
        fs.mkdirs(path, fsp);
        FileStatus fsStatus = fs.getFileStatus(path);
        LOG.info("Perms after creating " + fsStatus.getPermission().toShort()
            + ", Expected: " + fsp.toShort());
        if (fsStatus.getPermission().toShort() != fsp.toShort()) {
          LOG.info("Explicitly setting permissions to : " + fsp.toShort()
              + ", " + fsp);
          fs.setPermission(path, fsp);
        }
      } catch (FileAlreadyExistsException e) {
        LOG.info("Directory: [" + path + "] already exists.");
      }
    }
  }

  @Override
  protected void serviceStart() throws Exception {
    if (timelineClient != null) {
      timelineClient.start();
    }
    eventHandlingThread = new Thread(new Runnable() {
      @Override
      public void run() {
        JobHistoryEvent event = null;
        while (!stopped && !Thread.currentThread().isInterrupted()) {

          // Log the size of the history-event-queue every so often.
          if (eventCounter != 0 && eventCounter % 1000 == 0) {
            eventCounter = 0;
            LOG.info("Size of the JobHistory event queue is "
                + eventQueue.size());
          } else {
            eventCounter++;
          }

          try {
            event = eventQueue.take();
          } catch (InterruptedException e) {
            LOG.info("EventQueue take interrupted. Returning");
            return;
          }
          // If an event has been removed from the queue. Handle it.
          // The rest of the queue is handled via stop()
          // Clear the interrupt status if it's set before calling handleEvent
          // and set it if it was set before calling handleEvent. 
          // Interrupts received from other threads during handleEvent cannot be
          // dealth with - Shell.runCommand() ignores them.
          synchronized (lock) {
            boolean isInterrupted = Thread.interrupted();
            handleEvent(event);
            if (isInterrupted) {
                LOG.debug("Event handling interrupted");
                Thread.currentThread().interrupt();
              }
            }
          }
        }
    }, "eventHandlingThread");
    eventHandlingThread.start();
    super.serviceStart();
  }

  @Override
  protected void serviceStop() throws Exception {
    LOG.info("Stopping JobHistoryEventHandler. "
        + "Size of the outstanding queue size is " + eventQueue.size());
    stopped = true;
    //do not interrupt while event handling is in progress
    synchronized(lock) {
      if (eventHandlingThread != null) {
        LOG.debug("Interrupting Event Handling thread");
        eventHandlingThread.interrupt();
      } else {
        LOG.debug("Null event handling thread");
      }
    }

    try {
      if (eventHandlingThread != null) {
        LOG.debug("Waiting for Event Handling thread to complete");
        eventHandlingThread.join();
      }
    } catch (InterruptedException ie) {
      LOG.info("Interrupted Exception while stopping", ie);
    }

    // Cancel all timers - so that they aren't invoked during or after
    // the metaInfo object is wrapped up.
    for (MetaInfo mi : fileMap.values()) {
      try {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Shutting down timer for " + mi);
        }
        mi.shutDownTimer();
      } catch (IOException e) {
        LOG.info("Exception while cancelling delayed flush timer. "
            + "Likely caused by a failed flush " + e.getMessage());
      }
    }

    //write all the events remaining in queue
    Iterator<JobHistoryEvent> it = eventQueue.iterator();
    while(it.hasNext()) {
      JobHistoryEvent ev = it.next();
      LOG.info("In stop, writing event " + ev.getType());
      handleEvent(ev);
    }

    // Process JobUnsuccessfulCompletionEvent for jobIds which still haven't
    // closed their event writers
    if(forceJobCompletion) {
      for (Map.Entry<JobId,MetaInfo> jobIt : fileMap.entrySet()) {
        JobId toClose = jobIt.getKey();
        MetaInfo mi = jobIt.getValue();
        if(mi != null && mi.isWriterActive()) {
          LOG.warn("Found jobId " + toClose
            + " to have not been closed. Will close");
          //Create a JobFinishEvent so that it is written to the job history
          final Job job = context.getJob(toClose);
          JobUnsuccessfulCompletionEvent jucEvent =
            new JobUnsuccessfulCompletionEvent(TypeConverter.fromYarn(toClose),
                System.currentTimeMillis(), job.getCompletedMaps(),
                job.getCompletedReduces(),
                createJobStateForJobUnsuccessfulCompletionEvent(
                    mi.getForcedJobStateOnShutDown()),
                job.getDiagnostics());
          JobHistoryEvent jfEvent = new JobHistoryEvent(toClose, jucEvent);
          //Bypass the queue mechanism which might wait. Call the method directly
          handleEvent(jfEvent);
        }
      }
    }

    //close all file handles
    for (MetaInfo mi : fileMap.values()) {
      try {
        mi.closeWriter();
      } catch (IOException e) {
        LOG.info("Exception while closing file " + e.getMessage());
      }
    }
    if (timelineClient != null) {
      timelineClient.stop();
    }
    LOG.info("Stopped JobHistoryEventHandler. super.stop()");
    super.serviceStop();
  }

  protected EventWriter createEventWriter(Path historyFilePath)
      throws IOException {
    FSDataOutputStream out = stagingDirFS.create(historyFilePath, true);
    return new EventWriter(out, this.jhistMode);
  }
  
  /**
   * Create an event writer for the Job represented by the jobID.
   * Writes out the job configuration to the log directory.
   * This should be the first call to history for a job
   * 
   * @param jobId the jobId.
   * @param amStartedEvent
   * @throws IOException
   */
  protected void setupEventWriter(JobId jobId, AMStartedEvent amStartedEvent)
      throws IOException {
    if (stagingDirPath == null) {
      LOG.error("Log Directory is null, returning");
      throw new IOException("Missing Log Directory for History");
    }

    MetaInfo oldFi = fileMap.get(jobId);
    Configuration conf = getConfig();

    // TODO Ideally this should be written out to the job dir
    // (.staging/jobid/files - RecoveryService will need to be patched)
    Path historyFile = JobHistoryUtils.getStagingJobHistoryFile(
        stagingDirPath, jobId, startCount);
    String user = UserGroupInformation.getCurrentUser().getShortUserName();
    if (user == null) {
      throw new IOException(
          "User is null while setting up jobhistory eventwriter");
    }

    String jobName = context.getJob(jobId).getName();
    EventWriter writer = (oldFi == null) ? null : oldFi.writer;
 
    Path logDirConfPath =
        JobHistoryUtils.getStagingConfFile(stagingDirPath, jobId, startCount);
    if (writer == null) {
      try {
        writer = createEventWriter(historyFile);
        LOG.info("Event Writer setup for JobId: " + jobId + ", File: "
            + historyFile);
      } catch (IOException ioe) {
        LOG.info("Could not create log file: [" + historyFile + "] + for job "
            + "[" + jobName + "]");
        throw ioe;
      }
      
      //Write out conf only if the writer isn't already setup.
      if (conf != null) {
        // TODO Ideally this should be written out to the job dir
        // (.staging/jobid/files - RecoveryService will need to be patched)
        FSDataOutputStream jobFileOut = null;
        try {
          if (logDirConfPath != null) {
            jobFileOut = stagingDirFS.create(logDirConfPath, true);
            conf.writeXml(jobFileOut);
            jobFileOut.close();
          }
        } catch (IOException e) {
          LOG.info("Failed to write the job configuration file", e);
          throw e;
        }
      }
    }

    String queueName = JobConf.DEFAULT_QUEUE_NAME;
    if (conf != null) {
      queueName = conf.get(MRJobConfig.QUEUE_NAME, JobConf.DEFAULT_QUEUE_NAME);
    }

    MetaInfo fi = new MetaInfo(historyFile, logDirConfPath, writer,
        user, jobName, jobId, amStartedEvent.getForcedJobStateOnShutDown(),
        queueName);
    fi.getJobSummary().setJobId(jobId);
    fi.getJobSummary().setJobLaunchTime(amStartedEvent.getStartTime());
    fi.getJobSummary().setJobSubmitTime(amStartedEvent.getSubmitTime());
    fi.getJobIndexInfo().setJobStartTime(amStartedEvent.getStartTime());
    fi.getJobIndexInfo().setSubmitTime(amStartedEvent.getSubmitTime());
    fileMap.put(jobId, fi);
  }

  /** Close the event writer for this id 
   * @throws IOException */
  public void closeWriter(JobId id) throws IOException {
    try {
      final MetaInfo mi = fileMap.get(id);
      if (mi != null) {
        mi.closeWriter();
      }
      
    } catch (IOException e) {
      LOG.error("Error closing writer for JobID: " + id);
      throw e;
    }
  }

  @Override
  public void handle(JobHistoryEvent event) {
    try {
      if (isJobCompletionEvent(event.getHistoryEvent())) {
        // When the job is complete, flush slower but write faster.
        maxUnflushedCompletionEvents =
            maxUnflushedCompletionEvents * postJobCompletionMultiplier;
      }

      eventQueue.put(event);
    } catch (InterruptedException e) {
      throw new YarnRuntimeException(e);
    }
  }

  private boolean isJobCompletionEvent(HistoryEvent historyEvent) {
    if (EnumSet.of(EventType.JOB_FINISHED, EventType.JOB_FAILED,
        EventType.JOB_KILLED).contains(historyEvent.getEventType())) {
      return true;
    }
    return false;
  }

  @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 (timelineClient != null) {
          if (timelineServiceV2Enabled) {
            processEventForNewTimelineService(historyEvent, event.getJobID(),
                event.getTimestamp());
          } else {
            processEventForTimelineServer(historyEvent, event.getJobID(),
                event.getTimestamp());
          }
        }
        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.getFinishedMaps());
          mi.getJobIndexInfo().setNumReduces(
              jFinishedEvent.getFinishedReduces());
          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.getFinishedMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getFinishedReduces());
          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.getFinishedMaps());
          mi.getJobIndexInfo().setNumReduces(jucEvent.getFinishedReduces());
          mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
          closeEventWriter(event.getJobID());
          processDoneFiles(event.getJobID());
        } catch (IOException e) {
          throw new YarnRuntimeException(e);
        }
      }
    }
  }

  public void processEventForJobSummary(HistoryEvent event, JobSummary summary, 
      JobId jobId) {
    // context.getJob could be used for some of this info as well.
    switch (event.getEventType()) {
    case JOB_SUBMITTED:
      JobSubmittedEvent jse = (JobSubmittedEvent) event;
      summary.setUser(jse.getUserName());
      summary.setQueue(jse.getJobQueueName());
      summary.setJobSubmitTime(jse.getSubmitTime());
      summary.setJobName(jse.getJobName());
      break;
    case NORMALIZED_RESOURCE:
      NormalizedResourceEvent normalizedResourceEvent = 
            (NormalizedResourceEvent) event;
      if (normalizedResourceEvent.getTaskType() == TaskType.MAP) {
        summary.setResourcesPerMap((int) normalizedResourceEvent.getMemory());
      } else if (normalizedResourceEvent.getTaskType() == TaskType.REDUCE) {
        summary.setResourcesPerReduce((int) normalizedResourceEvent.getMemory());
      }
      break;  
    case JOB_INITED:
      JobInitedEvent jie = (JobInitedEvent) event;
      summary.setJobLaunchTime(jie.getLaunchTime());
      break;
    case MAP_ATTEMPT_STARTED:
      TaskAttemptStartedEvent mtase = (TaskAttemptStartedEvent) event;
      if (summary.getFirstMapTaskLaunchTime() == 0)
        summary.setFirstMapTaskLaunchTime(mtase.getStartTime());
      break;
    case REDUCE_ATTEMPT_STARTED:
      TaskAttemptStartedEvent rtase = (TaskAttemptStartedEvent) event;
      if (summary.getFirstReduceTaskLaunchTime() == 0)
        summary.setFirstReduceTaskLaunchTime(rtase.getStartTime());
      break;
    case JOB_FINISHED:
      JobFinishedEvent jfe = (JobFinishedEvent) event;
      summary.setJobFinishTime(jfe.getFinishTime());
      summary.setNumFinishedMaps(jfe.getFinishedMaps());
      summary.setNumFailedMaps(jfe.getFailedMaps());
      summary.setNumFinishedReduces(jfe.getFinishedReduces());
      summary.setNumFailedReduces(jfe.getFailedReduces());
      if (summary.getJobStatus() == null)
        summary
            .setJobStatus(org.apache.hadoop.mapreduce.JobStatus.State.SUCCEEDED
                .toString());
      // TODO JOB_FINISHED does not have state. Effectively job history does not
      // have state about the finished job.
      setSummarySlotSeconds(summary, jfe.getTotalCounters());
      break;
    case JOB_FAILED:
    case JOB_KILLED:
      JobUnsuccessfulCompletionEvent juce = (JobUnsuccessfulCompletionEvent) event;
      summary.setJobStatus(juce.getStatus());
      summary.setNumFinishedMaps(context.getJob(jobId).getTotalMaps());
      summary.setNumFinishedReduces(context.getJob(jobId).getTotalReduces());
      summary.setJobFinishTime(juce.getFinishTime());
      setSummarySlotSeconds(summary, context.getJob(jobId).getAllCounters());
      break;
    default:
      break;
    }
  }

  private void processEventForTimelineServer(HistoryEvent event, JobId jobId,
          long timestamp) {
    TimelineEvent tEvent = new TimelineEvent();
    tEvent.setEventType(StringUtils.toUpperCase(event.getEventType().name()));
    tEvent.setTimestamp(timestamp);
    TimelineEntity tEntity = new TimelineEntity();

    switch (event.getEventType()) {
      case JOB_SUBMITTED:
        JobSubmittedEvent jse =
            (JobSubmittedEvent) event;
        tEvent.addEventInfo("SUBMIT_TIME", jse.getSubmitTime());
        tEvent.addEventInfo("QUEUE_NAME", jse.getJobQueueName());
        tEvent.addEventInfo("JOB_NAME", jse.getJobName());
        tEvent.addEventInfo("USER_NAME", jse.getUserName());
        tEvent.addEventInfo("JOB_CONF_PATH", jse.getJobConfPath());
        tEvent.addEventInfo("ACLS", jse.getJobAcls());
        tEvent.addEventInfo("JOB_QUEUE_NAME", jse.getJobQueueName());
        tEvent.addEventInfo("WORKFLOW_ID", jse.getWorkflowId());
        tEvent.addEventInfo("WORKFLOW_NAME", jse.getWorkflowName());
        tEvent.addEventInfo("WORKFLOW_NAME_NAME", jse.getWorkflowNodeName());
        tEvent.addEventInfo("WORKFLOW_ADJACENCIES",
                jse.getWorkflowAdjacencies());
        tEvent.addEventInfo("WORKFLOW_TAGS", jse.getWorkflowTags());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_STATUS_CHANGED:
        JobStatusChangedEvent jsce = (JobStatusChangedEvent) event;
        tEvent.addEventInfo("STATUS", jsce.getStatus());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_INFO_CHANGED:
        JobInfoChangeEvent jice = (JobInfoChangeEvent) event;
        tEvent.addEventInfo("SUBMIT_TIME", jice.getSubmitTime());
        tEvent.addEventInfo("LAUNCH_TIME", jice.getLaunchTime());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_INITED:
        JobInitedEvent jie = (JobInitedEvent) event;
        tEvent.addEventInfo("START_TIME", jie.getLaunchTime());
        tEvent.addEventInfo("STATUS", jie.getStatus());
        tEvent.addEventInfo("TOTAL_MAPS", jie.getTotalMaps());
        tEvent.addEventInfo("TOTAL_REDUCES", jie.getTotalReduces());
        tEvent.addEventInfo("UBERIZED", jie.getUberized());
        tEntity.setStartTime(jie.getLaunchTime());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_PRIORITY_CHANGED:
        JobPriorityChangeEvent jpce = (JobPriorityChangeEvent) event;
        tEvent.addEventInfo("PRIORITY", jpce.getPriority().toString());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_QUEUE_CHANGED:
        JobQueueChangeEvent jqe = (JobQueueChangeEvent) event;
        tEvent.addEventInfo("QUEUE_NAMES", jqe.getJobQueueName());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_FAILED:
      case JOB_KILLED:
      case JOB_ERROR:
        JobUnsuccessfulCompletionEvent juce =
              (JobUnsuccessfulCompletionEvent) event;
        tEvent.addEventInfo("FINISH_TIME", juce.getFinishTime());
        tEvent.addEventInfo("NUM_MAPS", juce.getFinishedMaps());
        tEvent.addEventInfo("NUM_REDUCES", juce.getFinishedReduces());
        tEvent.addEventInfo("JOB_STATUS", juce.getStatus());
        tEvent.addEventInfo("DIAGNOSTICS", juce.getDiagnostics());
        tEvent.addEventInfo("FINISHED_MAPS", juce.getFinishedMaps());
        tEvent.addEventInfo("FINISHED_REDUCES", juce.getFinishedReduces());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case JOB_FINISHED:
        JobFinishedEvent jfe = (JobFinishedEvent) event;
        tEvent.addEventInfo("FINISH_TIME", jfe.getFinishTime());
        tEvent.addEventInfo("NUM_MAPS", jfe.getFinishedMaps());
        tEvent.addEventInfo("NUM_REDUCES", jfe.getFinishedReduces());
        tEvent.addEventInfo("FAILED_MAPS", jfe.getFailedMaps());
        tEvent.addEventInfo("FAILED_REDUCES", jfe.getFailedReduces());
        tEvent.addEventInfo("FINISHED_MAPS", jfe.getFinishedMaps());
        tEvent.addEventInfo("FINISHED_REDUCES", jfe.getFinishedReduces());
        tEvent.addEventInfo("MAP_COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(jfe.getMapCounters()));
        tEvent.addEventInfo("REDUCE_COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(jfe.getReduceCounters()));
        tEvent.addEventInfo("TOTAL_COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(jfe.getTotalCounters()));
        tEvent.addEventInfo("JOB_STATUS", JobState.SUCCEEDED.toString());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      case TASK_STARTED:
        TaskStartedEvent tse = (TaskStartedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tse.getTaskType().toString());
        tEvent.addEventInfo("START_TIME", tse.getStartTime());
        tEvent.addEventInfo("SPLIT_LOCATIONS", tse.getSplitLocations());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tse.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case TASK_FAILED:
        TaskFailedEvent tfe = (TaskFailedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tfe.getTaskType().toString());
        tEvent.addEventInfo("STATUS", TaskStatus.State.FAILED.toString());
        tEvent.addEventInfo("FINISH_TIME", tfe.getFinishTime());
        tEvent.addEventInfo("ERROR", tfe.getError());
        tEvent.addEventInfo("FAILED_ATTEMPT_ID",
                tfe.getFailedAttemptID() == null ?
                "" : tfe.getFailedAttemptID().toString());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(tfe.getCounters()));
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tfe.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case TASK_UPDATED:
        TaskUpdatedEvent tue = (TaskUpdatedEvent) event;
        tEvent.addEventInfo("FINISH_TIME", tue.getFinishTime());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tue.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case TASK_FINISHED:
        TaskFinishedEvent tfe2 = (TaskFinishedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tfe2.getTaskType().toString());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(tfe2.getCounters()));
        tEvent.addEventInfo("FINISH_TIME", tfe2.getFinishTime());
        tEvent.addEventInfo("STATUS", TaskStatus.State.SUCCEEDED.toString());
        tEvent.addEventInfo("SUCCESSFUL_TASK_ATTEMPT_ID",
            tfe2.getSuccessfulTaskAttemptId() == null ?
            "" : tfe2.getSuccessfulTaskAttemptId().toString());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tfe2.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case MAP_ATTEMPT_STARTED:
      case CLEANUP_ATTEMPT_STARTED:
      case REDUCE_ATTEMPT_STARTED:
      case SETUP_ATTEMPT_STARTED:
        TaskAttemptStartedEvent tase = (TaskAttemptStartedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tase.getTaskType().toString());
        tEvent.addEventInfo("TASK_ATTEMPT_ID",
            tase.getTaskAttemptId().toString());
        tEvent.addEventInfo("START_TIME", tase.getStartTime());
        tEvent.addEventInfo("HTTP_PORT", tase.getHttpPort());
        tEvent.addEventInfo("TRACKER_NAME", tase.getTrackerName());
        tEvent.addEventInfo("SHUFFLE_PORT", tase.getShufflePort());
        tEvent.addEventInfo("CONTAINER_ID", tase.getContainerId() == null ?
            "" : tase.getContainerId().toString());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tase.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case MAP_ATTEMPT_FAILED:
      case CLEANUP_ATTEMPT_FAILED:
      case REDUCE_ATTEMPT_FAILED:
      case SETUP_ATTEMPT_FAILED:
      case MAP_ATTEMPT_KILLED:
      case CLEANUP_ATTEMPT_KILLED:
      case REDUCE_ATTEMPT_KILLED:
      case SETUP_ATTEMPT_KILLED:
        TaskAttemptUnsuccessfulCompletionEvent tauce =
                (TaskAttemptUnsuccessfulCompletionEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tauce.getTaskType().toString());
        tEvent.addEventInfo("TASK_ATTEMPT_ID",
            tauce.getTaskAttemptId() == null ?
            "" : tauce.getTaskAttemptId().toString());
        tEvent.addEventInfo("FINISH_TIME", tauce.getFinishTime());
        tEvent.addEventInfo("ERROR", tauce.getError());
        tEvent.addEventInfo("STATUS", tauce.getTaskStatus());
        tEvent.addEventInfo("HOSTNAME", tauce.getHostname());
        tEvent.addEventInfo("PORT", tauce.getPort());
        tEvent.addEventInfo("RACK_NAME", tauce.getRackName());
        tEvent.addEventInfo("SHUFFLE_FINISH_TIME", tauce.getFinishTime());
        tEvent.addEventInfo("SORT_FINISH_TIME", tauce.getFinishTime());
        tEvent.addEventInfo("MAP_FINISH_TIME", tauce.getFinishTime());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(tauce.getCounters()));
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tauce.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case MAP_ATTEMPT_FINISHED:
        MapAttemptFinishedEvent mafe = (MapAttemptFinishedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", mafe.getTaskType().toString());
        tEvent.addEventInfo("FINISH_TIME", mafe.getFinishTime());
        tEvent.addEventInfo("STATUS", mafe.getTaskStatus());
        tEvent.addEventInfo("STATE", mafe.getState());
        tEvent.addEventInfo("MAP_FINISH_TIME", mafe.getMapFinishTime());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(mafe.getCounters()));
        tEvent.addEventInfo("HOSTNAME", mafe.getHostname());
        tEvent.addEventInfo("PORT", mafe.getPort());
        tEvent.addEventInfo("RACK_NAME", mafe.getRackName());
        tEvent.addEventInfo("ATTEMPT_ID", mafe.getAttemptId() == null ?
            "" : mafe.getAttemptId().toString());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(mafe.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case REDUCE_ATTEMPT_FINISHED:
        ReduceAttemptFinishedEvent rafe = (ReduceAttemptFinishedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", rafe.getTaskType().toString());
        tEvent.addEventInfo("ATTEMPT_ID", rafe.getAttemptId() == null ?
            "" : rafe.getAttemptId().toString());
        tEvent.addEventInfo("FINISH_TIME", rafe.getFinishTime());
        tEvent.addEventInfo("STATUS", rafe.getTaskStatus());
        tEvent.addEventInfo("STATE", rafe.getState());
        tEvent.addEventInfo("SHUFFLE_FINISH_TIME", rafe.getShuffleFinishTime());
        tEvent.addEventInfo("SORT_FINISH_TIME", rafe.getSortFinishTime());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(rafe.getCounters()));
        tEvent.addEventInfo("HOSTNAME", rafe.getHostname());
        tEvent.addEventInfo("PORT", rafe.getPort());
        tEvent.addEventInfo("RACK_NAME", rafe.getRackName());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(rafe.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case SETUP_ATTEMPT_FINISHED:
      case CLEANUP_ATTEMPT_FINISHED:
        TaskAttemptFinishedEvent tafe = (TaskAttemptFinishedEvent) event;
        tEvent.addEventInfo("TASK_TYPE", tafe.getTaskType().toString());
        tEvent.addEventInfo("ATTEMPT_ID", tafe.getAttemptId() == null ?
            "" : tafe.getAttemptId().toString());
        tEvent.addEventInfo("FINISH_TIME", tafe.getFinishTime());
        tEvent.addEventInfo("STATUS", tafe.getTaskStatus());
        tEvent.addEventInfo("STATE", tafe.getState());
        tEvent.addEventInfo("COUNTERS_GROUPS",
            JobHistoryEventUtils.countersToJSON(tafe.getCounters()));
        tEvent.addEventInfo("HOSTNAME", tafe.getHostname());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(tafe.getTaskId().toString());
        tEntity.setEntityType(MAPREDUCE_TASK_ENTITY_TYPE);
        tEntity.addRelatedEntity(MAPREDUCE_JOB_ENTITY_TYPE, jobId.toString());
        break;
      case AM_STARTED:
        AMStartedEvent ase = (AMStartedEvent) event;
        tEvent.addEventInfo("APPLICATION_ATTEMPT_ID",
                ase.getAppAttemptId() == null ?
                "" : ase.getAppAttemptId().toString());
        tEvent.addEventInfo("CONTAINER_ID", ase.getContainerId() == null ?
                "" : ase.getContainerId().toString());
        tEvent.addEventInfo("NODE_MANAGER_HOST", ase.getNodeManagerHost());
        tEvent.addEventInfo("NODE_MANAGER_PORT", ase.getNodeManagerPort());
        tEvent.addEventInfo("NODE_MANAGER_HTTP_PORT",
                ase.getNodeManagerHttpPort());
        tEvent.addEventInfo("START_TIME", ase.getStartTime());
        tEvent.addEventInfo("SUBMIT_TIME", ase.getSubmitTime());
        tEntity.addEvent(tEvent);
        tEntity.setEntityId(jobId.toString());
        tEntity.setEntityType(MAPREDUCE_JOB_ENTITY_TYPE);
        break;
      default:
        break;
    }

    try {
      TimelinePutResponse response = timelineClient.putEntities(tEntity);
      List<TimelinePutResponse.TimelinePutError> errors = response.getErrors();
      if (errors.size() == 0) {
        if (LOG.isDebugEnabled()) {
          LOG.debug("Timeline entities are successfully put in event " + event
              .getEventType());
        }
      } else {
        for (TimelinePutResponse.TimelinePutError error : errors) {
          LOG.error(
              "Error when publishing entity [" + error.getEntityType() + ","
                  + error.getEntityId() + "], server side error code: "
                  + error.getErrorCode());
        }
      }
    } catch (YarnException | IOException | ClientHandlerException ex) {
      LOG.error("Error putting entity " + tEntity.getEntityId() + " to Timeline"
          + "Server", ex);
    }
  }

  // create JobEntity from HistoryEvent with adding other info, like:
  // jobId, timestamp and entityType.
  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createJobEntity(HistoryEvent event, long timestamp, JobId jobId,
      String entityType, boolean setCreatedTime) {

    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
        createBaseEntity(event, timestamp, entityType, setCreatedTime);
    entity.setId(jobId.toString());
    return entity;
  }

  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createJobEntity(JobId jobId) {
    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
        new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
    entity.setId(jobId.toString());
    entity.setType(MAPREDUCE_JOB_ENTITY_TYPE);
    return entity;
  }

  // create ApplicationEntity with job finished Metrics from HistoryEvent
  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createAppEntityWithJobMetrics(HistoryEvent event, JobId jobId) {
    ApplicationEntity entity = new ApplicationEntity();
    entity.setId(jobId.getAppId().toString());
    entity.setMetrics(event.getTimelineMetrics());
    return entity;
  }

  // create BaseEntity from HistoryEvent with adding other info, like:
  // timestamp and entityType.
  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createBaseEntity(HistoryEvent event, long timestamp, String entityType,
      boolean setCreatedTime) {
    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEvent tEvent =
        event.toTimelineEvent();
    tEvent.setTimestamp(timestamp);

    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
        new org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity();
    entity.addEvent(tEvent);
    entity.setType(entityType);
    if (setCreatedTime) {
      entity.setCreatedTime(timestamp);
    }
    Set<TimelineMetric> timelineMetrics = event.getTimelineMetrics();
    if (timelineMetrics != null) {
      entity.setMetrics(timelineMetrics);
    }
    return entity;
  }

  // create TaskEntity from HistoryEvent with adding other info, like:
  // taskId, jobId, timestamp, entityType and relatedJobEntity.
  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createTaskEntity(HistoryEvent event, long timestamp, String taskId,
      String entityType, String relatedJobEntity, JobId jobId,
      boolean setCreatedTime) {
    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
        createBaseEntity(event, timestamp, entityType, setCreatedTime);
    entity.setId(taskId);
    if (event.getEventType() == EventType.TASK_STARTED) {
      entity.addInfo("TASK_TYPE",
          ((TaskStartedEvent)event).getTaskType().toString());
    }
    entity.addIsRelatedToEntity(relatedJobEntity, jobId.toString());
    return entity;
  }

  // create TaskAttemptEntity from HistoryEvent with adding other info, like:
  // timestamp, taskAttemptId, entityType, relatedTaskEntity and taskId.
  private org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
      createTaskAttemptEntity(HistoryEvent event, long timestamp,
      String taskAttemptId, String entityType, String relatedTaskEntity,
      String taskId, boolean setCreatedTime) {
    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity entity =
        createBaseEntity(event, timestamp, entityType, setCreatedTime);
    entity.setId(taskAttemptId);
    entity.addIsRelatedToEntity(relatedTaskEntity, taskId);
    return entity;
  }

  private void publishConfigsOnJobSubmittedEvent(JobSubmittedEvent event,
      JobId jobId) {
    if (event.getJobConf() == null) {
      return;
    }
    // Publish job configurations both as job and app entity.
    // Configs are split into multiple entities if they exceed 100kb in size.
    org.apache.hadoop.yarn.api.records.timelineservice.
        TimelineEntity jobEntityForConfigs = createJobEntity(jobId);
    ApplicationEntity appEntityForConfigs = new ApplicationEntity();
    String appId = jobId.getAppId().toString();
    appEntityForConfigs.setId(appId);
    try {
      int configSize = 0;
      for (Map.Entry<String, String> entry : event.getJobConf()) {
        int size = entry.getKey().length() + entry.getValue().length();
        configSize += size;
        if (configSize > JobHistoryEventUtils.ATS_CONFIG_PUBLISH_SIZE_BYTES) {
          if (jobEntityForConfigs.getConfigs().size() > 0) {
            timelineClient.putEntities(jobEntityForConfigs);
            timelineClient.putEntities(appEntityForConfigs);
            jobEntityForConfigs = createJobEntity(jobId);
            appEntityForConfigs = new ApplicationEntity();
            appEntityForConfigs.setId(appId);
          }
          configSize = size;
        }
        jobEntityForConfigs.addConfig(entry.getKey(), entry.getValue());
        appEntityForConfigs.addConfig(entry.getKey(), entry.getValue());
      }
      if (configSize > 0) {
        timelineClient.putEntities(jobEntityForConfigs);
        timelineClient.putEntities(appEntityForConfigs);
      }
    } catch (IOException | YarnException e) {
      LOG.error("Exception while publishing configs on JOB_SUBMITTED Event " +
          " for the job : " + jobId, e);
    }
  }

  private void processEventForNewTimelineService(HistoryEvent event,
      JobId jobId, long timestamp) {
    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity tEntity =
        null;
    String taskId = null;
    String taskAttemptId = null;
    boolean setCreatedTime = false;

    switch (event.getEventType()) {
    // Handle job events
    case JOB_SUBMITTED:
      setCreatedTime = true;
      break;
    case JOB_STATUS_CHANGED:
    case JOB_INFO_CHANGED:
    case JOB_INITED:
    case JOB_PRIORITY_CHANGED:
    case JOB_QUEUE_CHANGED:
    case JOB_FAILED:
    case JOB_KILLED:
    case JOB_ERROR:
    case JOB_FINISHED:
    case AM_STARTED:
    case NORMALIZED_RESOURCE:
      break;
    // Handle task events
    case TASK_STARTED:
      setCreatedTime = true;
      taskId = ((TaskStartedEvent)event).getTaskId().toString();
      break;
    case TASK_FAILED:
      taskId = ((TaskFailedEvent)event).getTaskId().toString();
      break;
    case TASK_UPDATED:
      taskId = ((TaskUpdatedEvent)event).getTaskId().toString();
      break;
    case TASK_FINISHED:
      taskId = ((TaskFinishedEvent)event).getTaskId().toString();
      break;
    case MAP_ATTEMPT_STARTED:
    case REDUCE_ATTEMPT_STARTED:
      setCreatedTime = true;
      taskId = ((TaskAttemptStartedEvent)event).getTaskId().toString();
      taskAttemptId = ((TaskAttemptStartedEvent)event).
          getTaskAttemptId().toString();
      break;
    case CLEANUP_ATTEMPT_STARTED:
    case SETUP_ATTEMPT_STARTED:
      taskId = ((TaskAttemptStartedEvent)event).getTaskId().toString();
      taskAttemptId = ((TaskAttemptStartedEvent)event).
          getTaskAttemptId().toString();
      break;
    case MAP_ATTEMPT_FAILED:
    case CLEANUP_ATTEMPT_FAILED:
    case REDUCE_ATTEMPT_FAILED:
    case SETUP_ATTEMPT_FAILED:
    case MAP_ATTEMPT_KILLED:
    case CLEANUP_ATTEMPT_KILLED:
    case REDUCE_ATTEMPT_KILLED:
    case SETUP_ATTEMPT_KILLED:
      taskId = ((TaskAttemptUnsuccessfulCompletionEvent)event).
          getTaskId().toString();
      taskAttemptId = ((TaskAttemptUnsuccessfulCompletionEvent)event).
          getTaskAttemptId().toString();
      break;
    case MAP_ATTEMPT_FINISHED:
      taskId = ((MapAttemptFinishedEvent)event).getTaskId().toString();
      taskAttemptId = ((MapAttemptFinishedEvent)event).
          getAttemptId().toString();
      break;
    case REDUCE_ATTEMPT_FINISHED:
      taskId = ((ReduceAttemptFinishedEvent)event).getTaskId().toString();
      taskAttemptId = ((ReduceAttemptFinishedEvent)event).
          getAttemptId().toString();
      break;
    case SETUP_ATTEMPT_FINISHED:
    case CLEANUP_ATTEMPT_FINISHED:
      taskId = ((TaskAttemptFinishedEvent)event).getTaskId().toString();
      taskAttemptId = ((TaskAttemptFinishedEvent)event).
          getAttemptId().toString();
      break;
    default:
      LOG.warn("EventType: " + event.getEventType() + " cannot be recognized" +
          " and handled by timeline service.");
      return;
    }

    org.apache.hadoop.yarn.api.records.timelineservice.TimelineEntity
        appEntityWithJobMetrics = null;
    if (taskId == null) {
      // JobEntity
      tEntity = createJobEntity(event, timestamp, jobId,
          MAPREDUCE_JOB_ENTITY_TYPE, setCreatedTime);
      if (event.getEventType() == EventType.JOB_FINISHED
          && event.getTimelineMetrics() != null) {
        appEntityWithJobMetrics = createAppEntityWithJobMetrics(event, jobId);
      }
    } else {
      if (taskAttemptId == null) {
        // TaskEntity
        tEntity = createTaskEntity(event, timestamp, taskId,
            MAPREDUCE_TASK_ENTITY_TYPE, MAPREDUCE_JOB_ENTITY_TYPE,
            jobId, setCreatedTime);
      } else {
        // TaskAttemptEntity
        tEntity = createTaskAttemptEntity(event, timestamp, taskAttemptId,
            MAPREDUCE_TASK_ATTEMPT_ENTITY_TYPE, MAPREDUCE_TASK_ENTITY_TYPE,
            taskId, setCreatedTime);
      }
    }
    try {
      if (appEntityWithJobMetrics == null) {
        timelineClient.putEntitiesAsync(tEntity);
      } else {
        timelineClient.putEntities(tEntity, appEntityWithJobMetrics);
      }
    } catch (IOException | YarnException e) {
      LOG.error("Failed to process Event " + event.getEventType()
          + " for the job : " + jobId, e);
      return;
    }
    if (event.getEventType() == EventType.JOB_SUBMITTED) {
      // Publish configs after main job submitted event has been posted.
      publishConfigsOnJobSubmittedEvent((JobSubmittedEvent)event, jobId);
    }
  }

  private void setSummarySlotSeconds(JobSummary summary, Counters allCounters) {

    Counter slotMillisMapCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_MAPS);
    if (slotMillisMapCounter != null) {
      summary.setMapSlotSeconds(slotMillisMapCounter.getValue() / 1000);
    }

    Counter slotMillisReduceCounter = allCounters
      .findCounter(JobCounter.SLOTS_MILLIS_REDUCES);
    if (slotMillisReduceCounter != null) {
      summary.setReduceSlotSeconds(slotMillisReduceCounter.getValue() / 1000);
    }
  }

  protected void closeEventWriter(JobId jobId) throws IOException {
    final MetaInfo mi = fileMap.get(jobId);
    if (mi == null) {
      throw new IOException("No MetaInfo found for JobId: [" + jobId + "]");
    }

    if (!mi.isWriterActive()) {
      throw new IOException(
          "Inactive Writer: Likely received multiple JobFinished / " +
          "JobUnsuccessful events for JobId: ["
              + jobId + "]");
    }

    // Close the Writer
    try {
      mi.closeWriter();
    } catch (IOException e) {
      LOG.error("Error closing writer for JobID: " + jobId);
      throw e;
    }
  }

  protected void processDoneFiles(JobId jobId) throws IOException {

    final MetaInfo mi = fileMap.get(jobId);
    if (mi == null) {
      throw new IOException("No MetaInfo found for JobId: [" + jobId + "]");
    }

    if (mi.getHistoryFile() == null) {
      LOG.warn("No file for job-history with " + jobId + " found in cache!");
    }
    if (mi.getConfFile() == null) {
      LOG.warn("No file for jobconf with " + jobId + " found in cache!");
    }
      
    // Writing out the summary file.
    // TODO JH enhancement - reuse this file to store additional indexing info
    // like ACLs, etc. JHServer can use HDFS append to build an index file
    // with more info than is available via the filename.
    Path qualifiedSummaryDoneFile = null;
    FSDataOutputStream summaryFileOut = null;
    try {
      String doneSummaryFileName = getTempFileName(JobHistoryUtils
          .getIntermediateSummaryFileName(jobId));
      qualifiedSummaryDoneFile = doneDirFS.makeQualified(new Path(
          doneDirPrefixPath, doneSummaryFileName));
      summaryFileOut = doneDirFS.create(qualifiedSummaryDoneFile, true);
      summaryFileOut.writeUTF(mi.getJobSummary().getJobSummaryString());
      summaryFileOut.close();
      doneDirFS.setPermission(qualifiedSummaryDoneFile, new FsPermission(
          JobHistoryUtils.HISTORY_INTERMEDIATE_FILE_PERMISSIONS));
    } catch (IOException e) {
      LOG.info("Unable to write out JobSummaryInfo to ["
          + qualifiedSummaryDoneFile + "]", e);
      throw e;
    }

    try {

      // Move historyFile to Done Folder.
      Path qualifiedDoneFile = null;
      if (mi.getHistoryFile() != null) {
        Path historyFile = mi.getHistoryFile();
        Path qualifiedLogFile = stagingDirFS.makeQualified(historyFile);
        int jobNameLimit =
            getConfig().getInt(JHAdminConfig.MR_HS_JOBNAME_LIMIT,
            JHAdminConfig.DEFAULT_MR_HS_JOBNAME_LIMIT);
        String doneJobHistoryFileName =
            getTempFileName(FileNameIndexUtils.getDoneFileName(mi
                .getJobIndexInfo(), jobNameLimit));
        qualifiedDoneFile =
            doneDirFS.makeQualified(new Path(doneDirPrefixPath,
                doneJobHistoryFileName));
        moveToDoneNow(qualifiedLogFile, qualifiedDoneFile);
      }

      // Move confFile to Done Folder
      Path qualifiedConfDoneFile = null;
      if (mi.getConfFile() != null) {
        Path confFile = mi.getConfFile();
        Path qualifiedConfFile = stagingDirFS.makeQualified(confFile);
        String doneConfFileName =
            getTempFileName(JobHistoryUtils
                .getIntermediateConfFileName(jobId));
        qualifiedConfDoneFile =
            doneDirFS.makeQualified(new Path(doneDirPrefixPath,
                doneConfFileName));
        moveToDoneNow(qualifiedConfFile, qualifiedConfDoneFile);
      }
      
      moveTmpToDone(qualifiedSummaryDoneFile);
      moveTmpToDone(qualifiedConfDoneFile);
      moveTmpToDone(qualifiedDoneFile);

    } catch (IOException e) {
      LOG.error("Error closing writer for JobID: " + jobId);
      throw e;
    }
  }

  private class FlushTimerTask extends TimerTask {
    private MetaInfo metaInfo;
    private IOException ioe = null;
    private volatile boolean shouldRun = true;

    FlushTimerTask(MetaInfo metaInfo) {
      this.metaInfo = metaInfo;
    }

    @Override
    public void run() {
      LOG.debug("In flush timer task");
      synchronized (lock) {
        try {
          if (!metaInfo.isTimerShutDown() && shouldRun)
            metaInfo.flush();
        } catch (IOException e) {
          ioe = e;
        }
      }
    }

    public IOException getException() {
      return ioe;
    }

    public void stop() {
      shouldRun = false;
      this.cancel();
    }
  }

  protected class MetaInfo {
    private Path historyFile;
    private Path confFile;
    private EventWriter writer;
    JobIndexInfo jobIndexInfo;
    JobSummary jobSummary;
    Timer flushTimer; 
    FlushTimerTask flushTimerTask;
    private boolean isTimerShutDown = false;
    private String forcedJobStateOnShutDown;

    MetaInfo(Path historyFile, Path conf, EventWriter writer, String user,
        String jobName, JobId jobId, String forcedJobStateOnShutDown,
        String queueName) {
      this.historyFile = historyFile;
      this.confFile = conf;
      this.writer = writer;
      this.jobIndexInfo =
          new JobIndexInfo(-1, -1, user, jobName, jobId, -1, -1, null,
                           queueName);
      this.jobSummary = new JobSummary();
      this.flushTimer = new Timer("FlushTimer", true);
      this.forcedJobStateOnShutDown = forcedJobStateOnShutDown;
    }

    Path getHistoryFile() {
      return historyFile;
    }

    Path getConfFile() {
      return confFile;
    }

    JobIndexInfo getJobIndexInfo() {
      return jobIndexInfo;
    }

    JobSummary getJobSummary() {
      return jobSummary;
    }

    boolean isWriterActive() {
      return writer != null;
    }
    
    boolean isTimerShutDown() {
      return isTimerShutDown;
    }

    String getForcedJobStateOnShutDown() {
      return forcedJobStateOnShutDown;
    }

    @Override
    public String toString() {
      return "Job MetaInfo for "+ jobSummary.getJobId()
             + " history file " + historyFile;
    }

    void closeWriter() throws IOException {
      LOG.debug("Closing Writer");
      synchronized (lock) {
        if (writer != null) {
          writer.close();
        }
        writer = null;
      }
    }

    void writeEvent(HistoryEvent event) throws IOException {
      LOG.debug("Writing event");
      synchronized (lock) {
        if (writer != null) {
          writer.write(event);
          processEventForFlush(event);
          maybeFlush(event);
        }
      }
    }

    void processEventForFlush(HistoryEvent historyEvent) throws IOException {
      if (EnumSet.of(EventType.MAP_ATTEMPT_FINISHED,
          EventType.MAP_ATTEMPT_FAILED, EventType.MAP_ATTEMPT_KILLED,
          EventType.REDUCE_ATTEMPT_FINISHED, EventType.REDUCE_ATTEMPT_FAILED,
          EventType.REDUCE_ATTEMPT_KILLED, EventType.TASK_FINISHED,
          EventType.TASK_FAILED, EventType.JOB_FINISHED, EventType.JOB_FAILED,
          EventType.JOB_KILLED).contains(historyEvent.getEventType())) {
        numUnflushedCompletionEvents++;
        if (!isTimerActive) {
          resetFlushTimer();
          if (!isTimerShutDown) {
            flushTimerTask = new FlushTimerTask(this);
            flushTimer.schedule(flushTimerTask, flushTimeout);
            isTimerActive = true;
          }
        }
      }
    }

    void resetFlushTimer() throws IOException {
      if (flushTimerTask != null) {
        IOException exception = flushTimerTask.getException();
        flushTimerTask.stop();
        if (exception != null) {
          throw exception;
        }
        flushTimerTask = null;
      }
      isTimerActive = false;
    }

    void maybeFlush(HistoryEvent historyEvent) throws IOException {
      if ((eventQueue.size() < minQueueSizeForBatchingFlushes 
          && numUnflushedCompletionEvents > 0)
          || numUnflushedCompletionEvents >= maxUnflushedCompletionEvents 
          || isJobCompletionEvent(historyEvent)) {
        this.flush();
      }
    }

    void flush() throws IOException {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Flushing " + toString());
      }
      synchronized (lock) {
        if (numUnflushedCompletionEvents != 0) { // skipped timer cancel.
          writer.flush();
          numUnflushedCompletionEvents = 0;
          resetFlushTimer();
        }
      }
    }

    void shutDownTimer() throws IOException {
      if (LOG.isDebugEnabled()) {
        LOG.debug("Shutting down timer "+ toString());
      }
      synchronized (lock) {
        isTimerShutDown = true;
        flushTimer.cancel();
        if (flushTimerTask != null && flushTimerTask.getException() != null) {
          throw flushTimerTask.getException();
        }
      }
    }
  }

  private void moveTmpToDone(Path tmpPath) throws IOException {
    if (tmpPath != null) {
      String tmpFileName = tmpPath.getName();
      String fileName = getFileNameFromTmpFN(tmpFileName);
      Path path = new Path(tmpPath.getParent(), fileName);
      doneDirFS.rename(tmpPath, path);
      LOG.info("Moved tmp to done: " + tmpPath + " to " + path);
    }
  }
  
  // TODO If the FS objects are the same, this should be a rename instead of a
  // copy.
  private void moveToDoneNow(Path fromPath, Path toPath) throws IOException {
    // check if path exists, in case of retries it may not exist
    if (stagingDirFS.exists(fromPath)) {
      LOG.info("Copying " + fromPath.toString() + " to " + toPath.toString());
      // TODO temporarily removing the existing dst
      if (doneDirFS.exists(toPath)) {
        doneDirFS.delete(toPath, true);
      }
      boolean copied = FileUtil.copy(stagingDirFS, fromPath, doneDirFS, toPath,
          false, getConfig());

      if (copied)
        LOG.info("Copied to done location: " + toPath);
      else 
        LOG.info("copy failed");
      doneDirFS.setPermission(toPath, new FsPermission(
          JobHistoryUtils.HISTORY_INTERMEDIATE_FILE_PERMISSIONS));
    }
  }

  boolean pathExists(FileSystem fileSys, Path path) throws IOException {
    return fileSys.exists(path);
  }

  private String getTempFileName(String srcFile) {
    return srcFile + "_tmp";
  }
  
  private String getFileNameFromTmpFN(String tmpFileName) {
    //TODO. Some error checking here.
    return tmpFileName.substring(0, tmpFileName.length()-4);
  }

  public void setForcejobCompletion(boolean forceJobCompletion) {
    this.forceJobCompletion = forceJobCompletion;
    LOG.info("JobHistoryEventHandler notified that forceJobCompletion is "
      + forceJobCompletion);
  }

  private String createJobStateForJobUnsuccessfulCompletionEvent(
      String forcedJobStateOnShutDown) {
    if (forcedJobStateOnShutDown == null || forcedJobStateOnShutDown
        .isEmpty()) {
      return JobState.KILLED.toString();
    } else if (forcedJobStateOnShutDown.equals(
        JobStateInternal.ERROR.toString()) ||
        forcedJobStateOnShutDown.equals(JobStateInternal.FAILED.toString())) {
      return JobState.FAILED.toString();
    } else if (forcedJobStateOnShutDown.equals(JobStateInternal.SUCCEEDED
        .toString())) {
      return JobState.SUCCEEDED.toString();
    }
    return JobState.KILLED.toString();
  }

  @VisibleForTesting
  boolean getFlushTimerStatus() {
    return isTimerActive;
  }
}