/**
 * 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.tez.dag.app;

import java.io.EOFException;
import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.yarn.api.records.LocalResource;
import org.apache.tez.common.TezCommonUtils;
import org.apache.tez.dag.api.TezConfiguration;
import org.apache.tez.dag.api.TezConstants;
import org.apache.tez.dag.app.dag.DAGState;
import org.apache.tez.dag.app.dag.Task;
import org.apache.tez.dag.app.dag.Vertex;
import org.apache.tez.dag.app.dag.impl.DAGImpl;
import org.apache.tez.dag.history.HistoryEvent;
import org.apache.tez.dag.history.HistoryEventType;
import org.apache.tez.dag.history.events.AMLaunchedEvent;
import org.apache.tez.dag.history.events.AMStartedEvent;
import org.apache.tez.dag.history.events.ContainerLaunchedEvent;
import org.apache.tez.dag.history.events.ContainerStoppedEvent;
import org.apache.tez.dag.history.events.DAGCommitStartedEvent;
import org.apache.tez.dag.history.events.DAGFinishedEvent;
import org.apache.tez.dag.history.events.DAGInitializedEvent;
import org.apache.tez.dag.history.events.DAGKillRequestEvent;
import org.apache.tez.dag.history.events.DAGStartedEvent;
import org.apache.tez.dag.history.events.DAGSubmittedEvent;
import org.apache.tez.dag.history.events.TaskAttemptFinishedEvent;
import org.apache.tez.dag.history.events.TaskAttemptStartedEvent;
import org.apache.tez.dag.history.events.TaskFinishedEvent;
import org.apache.tez.dag.history.events.TaskStartedEvent;
import org.apache.tez.dag.history.events.VertexCommitStartedEvent;
import org.apache.tez.dag.history.events.VertexRecoverableEventsGeneratedEvent;
import org.apache.tez.dag.history.events.VertexFinishedEvent;
import org.apache.tez.dag.history.events.VertexGroupCommitFinishedEvent;
import org.apache.tez.dag.history.events.VertexGroupCommitStartedEvent;
import org.apache.tez.dag.history.events.VertexInitializedEvent;
import org.apache.tez.dag.history.events.VertexParallelismUpdatedEvent;
import org.apache.tez.dag.history.events.VertexStartedEvent;
import org.apache.tez.dag.history.recovery.RecoveryService;
import org.apache.tez.dag.records.TezDAGID;
import org.apache.tez.dag.records.TezVertexID;
import org.apache.tez.dag.recovery.records.RecoveryProtos;
import org.apache.tez.dag.recovery.records.RecoveryProtos.SummaryEventProto;

import com.google.common.annotations.VisibleForTesting;

public class RecoveryParser {

  private static final Logger LOG = LoggerFactory.getLogger(RecoveryParser.class);

  private final DAGAppMaster dagAppMaster;
  private final FileSystem recoveryFS;
  private final Path recoveryDataDir;
  private final Path currentAttemptRecoveryDataDir;
  private final int recoveryBufferSize;
  private final int currentAttemptId;

  public RecoveryParser(DAGAppMaster dagAppMaster,
      FileSystem recoveryFS,
      Path recoveryDataDir,
      int currentAttemptId) throws IOException {
    this.dagAppMaster = dagAppMaster;
    this.recoveryFS = recoveryFS;
    this.recoveryDataDir = recoveryDataDir;
    this.currentAttemptId = currentAttemptId;
    this.currentAttemptRecoveryDataDir = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir,
        currentAttemptId);
    recoveryBufferSize = dagAppMaster.getConfig().getInt(
        TezConfiguration.DAG_RECOVERY_FILE_IO_BUFFER_SIZE,
        TezConfiguration.DAG_RECOVERY_FILE_IO_BUFFER_SIZE_DEFAULT);
    this.recoveryFS.mkdirs(currentAttemptRecoveryDataDir);
  }

  public static class RecoveredDAGData {
    public TezDAGID recoveredDagID = null;
    public DAGImpl recoveredDAG = null;
    public DAGState dagState = null;
    public boolean isCompleted = false;
    public boolean nonRecoverable = false;
    public boolean isSessionStopped = false;
    public String reason = null;
    public Map<String, LocalResource> cumulativeAdditionalResources = null;
  }

  private static void parseSummaryFile(FSDataInputStream inputStream)
      throws IOException {
    while (true) {
      RecoveryProtos.SummaryEventProto proto =
          RecoveryProtos.SummaryEventProto.parseDelimitedFrom(inputStream);
      if (proto == null) {
        LOG.info("Reached end of summary stream");
        break;
      }
      LOG.info("[SUMMARY]"
          + " dagId=" + proto.getDagId()
          + ", timestamp=" + proto.getTimestamp()
          + ", event=" + HistoryEventType.values()[proto.getEventType()]);
    }
  }

  private static HistoryEvent getNextEvent(FSDataInputStream inputStream)
      throws IOException {
    int eventTypeOrdinal = -1;
    try {
      eventTypeOrdinal = inputStream.readInt();
    } catch (EOFException eof) {
      return null;
    }
    if (eventTypeOrdinal < 0 || eventTypeOrdinal >=
        HistoryEventType.values().length) {
      // Corrupt data
      // reached end
      throw new IOException("Corrupt data found when trying to read next event type"
          + ", eventTypeOrdinal=" + eventTypeOrdinal);
    }
    HistoryEventType eventType = HistoryEventType.values()[eventTypeOrdinal];
    HistoryEvent event;
    switch (eventType) {
      case AM_LAUNCHED:
        event = new AMLaunchedEvent();
        break;
      case AM_STARTED:
        event = new AMStartedEvent();
        break;
      case DAG_SUBMITTED:
        event = new DAGSubmittedEvent();
        break;
      case DAG_INITIALIZED:
        event = new DAGInitializedEvent();
        break;
      case DAG_STARTED:
        event = new DAGStartedEvent();
        break;
      case DAG_COMMIT_STARTED:
        event = new DAGCommitStartedEvent();
        break;
      case DAG_FINISHED:
        event = new DAGFinishedEvent();
        break;
      case DAG_KILL_REQUEST:
        event = new DAGKillRequestEvent();
        break;
      case CONTAINER_LAUNCHED:
        event = new ContainerLaunchedEvent();
        break;
      case CONTAINER_STOPPED:
        event = new ContainerStoppedEvent();
        break;
      case VERTEX_INITIALIZED:
        event = new VertexInitializedEvent();
        break;
      case VERTEX_STARTED:
        event = new VertexStartedEvent();
        break;
      case VERTEX_PARALLELISM_UPDATED:
        event = new VertexParallelismUpdatedEvent();
        break;
      case VERTEX_COMMIT_STARTED:
        event = new VertexCommitStartedEvent();
        break;
      case VERTEX_GROUP_COMMIT_STARTED:
        event = new VertexGroupCommitStartedEvent();
        break;
      case VERTEX_GROUP_COMMIT_FINISHED:
        event = new VertexGroupCommitFinishedEvent();
        break;
      case VERTEX_FINISHED:
        event = new VertexFinishedEvent();
        break;
      case TASK_STARTED:
        event = new TaskStartedEvent();
        break;
      case TASK_FINISHED:
        event = new TaskFinishedEvent();
        break;
      case TASK_ATTEMPT_STARTED:
        event = new TaskAttemptStartedEvent();
        break;
      case TASK_ATTEMPT_FINISHED:
        event = new TaskAttemptFinishedEvent();
        break;
      case VERTEX_DATA_MOVEMENT_EVENTS_GENERATED:
        event = new VertexRecoverableEventsGeneratedEvent();
        break;
      default:
        throw new IOException("Invalid data found, unknown event type "
            + eventType);

    }
    if (LOG.isDebugEnabled()) {
      LOG.debug("Parsing event from input stream"
          + ", eventType=" + eventType);
    }
    try {
      event.fromProtoStream(inputStream);
    } catch (EOFException eof) {
      return null;
    }
    if (LOG.isDebugEnabled()) {
      LOG.debug("Parsed event from input stream"
          + ", eventType=" + eventType
          + ", event=" + event.toString());
    }
    return event;
  }


  public static List<HistoryEvent> parseDAGRecoveryFile(FSDataInputStream inputStream)
      throws IOException {
    List<HistoryEvent> historyEvents = new ArrayList<HistoryEvent>();
    while (true) {
      HistoryEvent historyEvent = getNextEvent(inputStream);
      if (historyEvent == null) {
        LOG.info("Reached end of stream");
        break;
      }
      historyEvents.add(historyEvent);
    }
    return historyEvents;
  }

  public static void main(String argv[]) throws IOException {
    // TODO clean up with better usage and error handling
    Configuration conf = new Configuration();
    String summaryPath = argv[0];
    List<String> dagPaths = new ArrayList<String>();
    if (argv.length > 1) {
      for (int i = 1; i < argv.length; ++i) {
        dagPaths.add(argv[i]);
      }
    }
    FileSystem fs = FileSystem.get(conf);
    LOG.info("Parsing Summary file " + summaryPath);
    parseSummaryFile(fs.open(new Path(summaryPath)));
    for (String dagPath : dagPaths) {
      LOG.info("Parsing DAG recovery file " + dagPath);
      List<HistoryEvent> historyEvents = parseDAGRecoveryFile(fs.open(new Path(dagPath)));
      for (HistoryEvent historyEvent : historyEvents) {
        LOG.info("Parsed event from recovery stream"
            + ", eventType=" + historyEvent.getEventType()
            + ", event=" + historyEvent);
      }
    }
  }

  private Path getSummaryPath(Path attemptRrecoveryDataDir) {
    return TezCommonUtils.getSummaryRecoveryPath(attemptRrecoveryDataDir);
  }

  private FSDataInputStream getSummaryStream(Path summaryPath)
      throws IOException {
    if (!recoveryFS.exists(summaryPath)) {
      return null;
    }
    return recoveryFS.open(summaryPath, recoveryBufferSize);
  }

  private Path getDAGRecoveryFilePath(Path recoveryDataDir,
      TezDAGID dagID) {
    return new Path(recoveryDataDir,
        dagID.toString() + TezConstants.DAG_RECOVERY_RECOVER_FILE_SUFFIX);
  }

  @VisibleForTesting
  DAGSummaryData getLastCompletedOrInProgressDAG(
      Map<TezDAGID, DAGSummaryData> dagSummaryDataMap) {
    DAGSummaryData inProgressDAG = null;
    DAGSummaryData lastCompletedDAG = null;
    for (Map.Entry<TezDAGID, DAGSummaryData> entry : dagSummaryDataMap.entrySet()) {
      if (!entry.getValue().completed) {
        if (inProgressDAG != null) {
          throw new RuntimeException("Multiple in progress DAGs seen"
              + ", dagId=" + inProgressDAG.dagId
              + ", dagId=" + entry.getKey());
        }
        inProgressDAG = entry.getValue();
      } else {
        if (lastCompletedDAG == null ||
            lastCompletedDAG.dagId.getId() < entry.getValue().dagId.getId()) {
          lastCompletedDAG = entry.getValue();
        }
      }
    }
    if (inProgressDAG == null) {
      return lastCompletedDAG;
    }
    return inProgressDAG;
  }

  @VisibleForTesting
  static class DAGSummaryData {

    final TezDAGID dagId;
    boolean completed = false;
    boolean dagCommitCompleted = true;
    DAGState dagState;
    Map<TezVertexID, Boolean> vertexCommitStatus =
        new HashMap<TezVertexID, Boolean>();
    Map<String, Boolean> vertexGroupCommitStatus =
        new HashMap<String, Boolean>();
    List<HistoryEvent> bufferedSummaryEvents =
        new ArrayList<HistoryEvent>();

    DAGSummaryData(TezDAGID dagId) {
      this.dagId = dagId;
    }

    void handleSummaryEvent(SummaryEventProto proto) throws IOException {
      HistoryEventType eventType =
          HistoryEventType.values()[proto.getEventType()];
      switch (eventType) {
        case DAG_SUBMITTED:
          completed = false;
          DAGSubmittedEvent dagSubmittedEvent = new DAGSubmittedEvent();
          dagSubmittedEvent.fromSummaryProtoStream(proto);
          break;
        case DAG_FINISHED:
          completed = true;
          dagCommitCompleted = true;
          DAGFinishedEvent dagFinishedEvent = new DAGFinishedEvent();
          dagFinishedEvent.fromSummaryProtoStream(proto);
          dagState = dagFinishedEvent.getState();
          break;
        case DAG_KILL_REQUEST:
          DAGKillRequestEvent killRequestEvent = new DAGKillRequestEvent();
          killRequestEvent.fromSummaryProtoStream(proto);
          bufferedSummaryEvents.add(killRequestEvent);
          break;
        case DAG_COMMIT_STARTED:
          dagCommitCompleted = false;
          break;
        case VERTEX_COMMIT_STARTED:
          VertexCommitStartedEvent vertexCommitStartedEvent =
              new VertexCommitStartedEvent();
          vertexCommitStartedEvent.fromSummaryProtoStream(proto);
          vertexCommitStatus.put(
              vertexCommitStartedEvent.getVertexID(), false);
          break;
        case VERTEX_FINISHED:
          VertexFinishedEvent vertexFinishedEvent =
              new VertexFinishedEvent();
          vertexFinishedEvent.fromSummaryProtoStream(proto);
          if (vertexCommitStatus.containsKey(vertexFinishedEvent.getVertexID())) {
            vertexCommitStatus.put(
                vertexFinishedEvent.getVertexID(), true);
            bufferedSummaryEvents.add(vertexFinishedEvent);
          }
          break;
        case VERTEX_GROUP_COMMIT_STARTED:
          VertexGroupCommitStartedEvent vertexGroupCommitStartedEvent =
              new VertexGroupCommitStartedEvent();
          vertexGroupCommitStartedEvent.fromSummaryProtoStream(proto);
          bufferedSummaryEvents.add(vertexGroupCommitStartedEvent);
          vertexGroupCommitStatus.put(
              vertexGroupCommitStartedEvent.getVertexGroupName(), false);
          break;
        case VERTEX_GROUP_COMMIT_FINISHED:
          VertexGroupCommitFinishedEvent vertexGroupCommitFinishedEvent =
              new VertexGroupCommitFinishedEvent();
          vertexGroupCommitFinishedEvent.fromSummaryProtoStream(proto);
          bufferedSummaryEvents.add(vertexGroupCommitFinishedEvent);
          vertexGroupCommitStatus.put(
              vertexGroupCommitFinishedEvent.getVertexGroupName(), true);
          break;
        default:
          String message = "Found invalid summary event that was not handled"
              + ", eventType=" + eventType.name();
          throw new IOException(message);
      }
    }

    @Override
    public String toString() {
      StringBuilder sb = new StringBuilder();
      sb.append("dagId=").append(dagId);
      sb.append(", dagCompleted=").append(completed);
      if (!vertexCommitStatus.isEmpty()) {
        sb.append(", vertexCommitStatuses=[");
        for (Entry<TezVertexID, Boolean> entry : vertexCommitStatus.entrySet()) {
          sb.append("{ vertexId=").append(entry.getKey())
              .append(", committed=").append(entry.getValue()).append("}, ");
        }
        sb.append("]");
      }
      if (!vertexGroupCommitStatus.isEmpty()) {
        sb.append(", vertexGroupCommitStatuses=[");
        for (Entry<String, Boolean> entry : vertexGroupCommitStatus.entrySet()) {
          sb.append("{ vertexGroup=").append(entry.getKey())
              .append(", committed=").append(entry.getValue()).append("}, ");
        }
        sb.append("]");
      }
      return sb.toString();
    }
  }

  private String isDAGRecoverable(DAGSummaryData data) {
    if (!data.dagCommitCompleted) {
      return "DAG Commit was in progress, not recoverable"
          + ", dagId=" + data.dagId;
    }
    if (!data.vertexCommitStatus.isEmpty()) {
      for (Entry<TezVertexID, Boolean> entry : data.vertexCommitStatus.entrySet()) {
        if (!(entry.getValue().booleanValue())) {
          return "Vertex Commit was in progress, not recoverable"
              + ", dagId=" + data.dagId
              + ", vertexId=" + entry.getKey();
        }
      }
    }
    if (!data.vertexGroupCommitStatus.isEmpty()) {
      for (Entry<String, Boolean> entry : data.vertexGroupCommitStatus.entrySet()) {
        if (!(entry.getValue().booleanValue())) {
          return "Vertex Group Commit was in progress, not recoverable"
              + ", dagId=" + data.dagId
              + ", vertexGroup=" + entry.getKey();
        }
      }
    }
    return null;
  }

  private List<Path> getSummaryFiles() throws IOException {
    List<Path> summaryFiles = new ArrayList<Path>();
    for (int i = 1; i < currentAttemptId; ++i) {
      Path attemptPath = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir, i);
      Path fatalErrorOccurred = new Path(attemptPath,
          RecoveryService.RECOVERY_FATAL_OCCURRED_DIR);
      if (recoveryFS.exists(fatalErrorOccurred)) {
        throw new IOException("Found that a fatal error occurred in"
            + " recovery during previous attempt, foundFile="
            + fatalErrorOccurred.toString());
      }
      Path summaryFile = getSummaryPath(attemptPath);
      if (recoveryFS.exists(summaryFile)) {
        summaryFiles.add(summaryFile);
      }
    }
    return summaryFiles;
  }

  private List<Path> getDAGRecoveryFiles(TezDAGID dagId) throws IOException {
    List<Path> recoveryFiles = new ArrayList<Path>();
    for (int i = 1; i < currentAttemptId; ++i) {
      Path attemptPath = TezCommonUtils.getAttemptRecoveryPath(recoveryDataDir, i);
      Path recoveryFile = getDAGRecoveryFilePath(attemptPath, dagId);
      if (recoveryFS.exists(recoveryFile)) {
        recoveryFiles.add(recoveryFile);
      }
    }
    return recoveryFiles;
  }

  public RecoveredDAGData parseRecoveryData() throws IOException {
    int dagCounter = 0;
    Map<TezDAGID, DAGSummaryData> dagSummaryDataMap =
        new HashMap<TezDAGID, DAGSummaryData>();
    List<Path> summaryFiles = getSummaryFiles();
    for (Path summaryFile : summaryFiles) {
      FileStatus summaryFileStatus = recoveryFS.getFileStatus(summaryFile);
      LOG.info("Parsing summary file"
          + ", path=" + summaryFile.toString()
          + ", len=" + summaryFileStatus.getLen()
          + ", lastModTime=" + summaryFileStatus.getModificationTime());
      FSDataInputStream summaryStream = getSummaryStream(
          summaryFile);
      while (true) {
        RecoveryProtos.SummaryEventProto proto;
        try {
          proto = RecoveryProtos.SummaryEventProto.parseDelimitedFrom(summaryStream);
          if (proto == null) {
            LOG.info("Reached end of summary stream");
            break;
          }
        } catch (EOFException eof) {
          LOG.info("Reached end of summary stream");
          break;
        }
        HistoryEventType eventType =
            HistoryEventType.values()[proto.getEventType()];
        if (LOG.isDebugEnabled()) {
          LOG.debug("[RECOVERY SUMMARY]"
              + " dagId=" + proto.getDagId()
              + ", timestamp=" + proto.getTimestamp()
              + ", event=" + eventType);
        }
        TezDAGID dagId = TezDAGID.fromString(proto.getDagId());
        if (dagId == null) {
          throw new IOException("null dagId, summary records may be corrupted");
        }
        if (dagCounter < dagId.getId()) {
          dagCounter = dagId.getId();
        }
        if (!dagSummaryDataMap.containsKey(dagId)) {
          dagSummaryDataMap.put(dagId, new DAGSummaryData(dagId));
        }
        try {
          dagSummaryDataMap.get(dagId).handleSummaryEvent(proto);
        } catch (Exception e) {
          // any exception when parsing protobuf
          throw new IOException("Error when parsing summary event proto", e);
        }
      }
      summaryStream.close();
    }

    // Set counter for next set of DAGs & update dagNames Set in DAGAppMaster
    dagAppMaster.setDAGCounter(dagCounter);
    for (DAGSummaryData dagSummaryData: dagSummaryDataMap.values()){
      dagAppMaster.dagIDs.add(dagSummaryData.dagId.toString());
    }

    DAGSummaryData lastInProgressDAGData =
        getLastCompletedOrInProgressDAG(dagSummaryDataMap);
    if (lastInProgressDAGData == null) {
      LOG.info("Nothing to recover as no uncompleted/completed DAGs found");
      return null;
    }
    TezDAGID lastInProgressDAG = lastInProgressDAGData.dagId;
    if (lastInProgressDAG == null) {
      LOG.info("Nothing to recover as no uncompleted/completed DAGs found");
      return null;
    }

    LOG.info("Checking if DAG is in recoverable state"
        + ", dagId=" + lastInProgressDAGData.dagId);

    final RecoveredDAGData recoveredDAGData = new RecoveredDAGData();
    if (lastInProgressDAGData.completed) {
      recoveredDAGData.isCompleted = true;
      recoveredDAGData.dagState = lastInProgressDAGData.dagState;
    }

    String nonRecoverableReason = isDAGRecoverable(lastInProgressDAGData);
    if (nonRecoverableReason != null) {
      LOG.warn("Found last inProgress DAG but not recoverable: "
          + lastInProgressDAGData);
      recoveredDAGData.nonRecoverable = true;
      recoveredDAGData.reason = nonRecoverableReason;
    }

    List<Path> dagRecoveryFiles = getDAGRecoveryFiles(lastInProgressDAG);
    boolean skipAllOtherEvents = false;
    Path lastRecoveryFile = null;
    for (Path dagRecoveryFile : dagRecoveryFiles) {
      if (skipAllOtherEvents) {
        LOG.warn("Other recovery files will be skipped due to error in the previous recovery file"
            + lastRecoveryFile);
        break;
      }
      lastRecoveryFile = dagRecoveryFile;
      LOG.info("Trying to recover dag from recovery file"
          + ", dagId=" + lastInProgressDAG.toString()
          + ", dagRecoveryFile=" + dagRecoveryFile);
      FSDataInputStream dagRecoveryStream = recoveryFS.open(dagRecoveryFile, recoveryBufferSize);
      while (true) {
        HistoryEvent event;
        try {
          event = getNextEvent(dagRecoveryStream);
          if (event == null) {
            LOG.info("Reached end of dag recovery stream");
            break;
          }
        } catch (EOFException eof) {
          LOG.info("Reached end of dag recovery stream");
          break;
        } catch (IOException ioe) {
          LOG.warn("Corrupt data found when trying to read next event", ioe);
          break;
        }
        if (skipAllOtherEvents) {
          // hit an error - skip reading other events
          break;
        }
        HistoryEventType eventType = event.getEventType();
        switch (eventType) {
          case DAG_SUBMITTED:
          {
            DAGSubmittedEvent submittedEvent = (DAGSubmittedEvent) event;
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            recoveredDAGData.recoveredDAG = dagAppMaster.createDAG(submittedEvent.getDAGPlan(),
                lastInProgressDAG);
            recoveredDAGData.cumulativeAdditionalResources = submittedEvent
              .getCumulativeAdditionalLocalResources();
            recoveredDAGData.recoveredDagID = recoveredDAGData.recoveredDAG.getID();
            dagAppMaster.setCurrentDAG(recoveredDAGData.recoveredDAG);
            if (recoveredDAGData.nonRecoverable) {
              skipAllOtherEvents = true;
            }
            break;
          }
          case DAG_INITIALIZED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            break;
          }
          case DAG_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            break;
          }
          case DAG_COMMIT_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            break;
          }
          case VERTEX_GROUP_COMMIT_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            break;
          }
          case VERTEX_GROUP_COMMIT_FINISHED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            break;
          }
          case DAG_KILL_REQUEST:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            break;
          }
          case DAG_FINISHED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            // If this is seen, nothing to recover
            assert recoveredDAGData.recoveredDAG != null;
            recoveredDAGData.recoveredDAG.restoreFromEvent(event);
            recoveredDAGData.isCompleted = true;
            recoveredDAGData.dagState =
                ((DAGFinishedEvent) event).getState();
            skipAllOtherEvents = true;
            break;
          }
          case CONTAINER_LAUNCHED:
          {
            // Nothing to do for now
            break;
          }
          case VERTEX_INITIALIZED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexInitializedEvent vEvent = (VertexInitializedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          case VERTEX_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexStartedEvent vEvent = (VertexStartedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          case VERTEX_PARALLELISM_UPDATED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexParallelismUpdatedEvent vEvent = (VertexParallelismUpdatedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          case VERTEX_COMMIT_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexCommitStartedEvent vEvent = (VertexCommitStartedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          case VERTEX_FINISHED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexFinishedEvent vEvent = (VertexFinishedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          case TASK_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            TaskStartedEvent tEvent = (TaskStartedEvent) event;
            Task task = recoveredDAGData.recoveredDAG.getVertex(
                tEvent.getTaskID().getVertexID()).getTask(tEvent.getTaskID());
            task.restoreFromEvent(tEvent);
            break;
          }
          case TASK_FINISHED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            TaskFinishedEvent tEvent = (TaskFinishedEvent) event;
            Task task = recoveredDAGData.recoveredDAG.getVertex(
                tEvent.getTaskID().getVertexID()).getTask(tEvent.getTaskID());
            task.restoreFromEvent(tEvent);
            break;
          }
          case TASK_ATTEMPT_STARTED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            TaskAttemptStartedEvent tEvent = (TaskAttemptStartedEvent) event;
            Task task =
                recoveredDAGData.recoveredDAG.getVertex(
                    tEvent.getTaskAttemptID().getTaskID().getVertexID())
                        .getTask(tEvent.getTaskAttemptID().getTaskID());
            task.restoreFromEvent(tEvent);
            break;
          }
          case TASK_ATTEMPT_FINISHED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            TaskAttemptFinishedEvent tEvent = (TaskAttemptFinishedEvent) event;
            Task task =
                recoveredDAGData.recoveredDAG.getVertex(
                    tEvent.getTaskAttemptID().getTaskID().getVertexID())
                    .getTask(tEvent.getTaskAttemptID().getTaskID());
            task.restoreFromEvent(tEvent);
            break;
          }
          case VERTEX_DATA_MOVEMENT_EVENTS_GENERATED:
          {
            LOG.info("Recovering from event"
                + ", eventType=" + eventType
                + ", event=" + event.toString());
            assert recoveredDAGData.recoveredDAG != null;
            VertexRecoverableEventsGeneratedEvent vEvent =
                (VertexRecoverableEventsGeneratedEvent) event;
            Vertex v = recoveredDAGData.recoveredDAG.getVertex(vEvent.getVertexID());
            v.restoreFromEvent(vEvent);
            break;
          }
          default:
            throw new RuntimeException("Invalid data found, unknown event type "
                + eventType);
        }
        if (LOG.isDebugEnabled()) {
          LOG.debug("[DAG RECOVERY]"
              + " dagId=" + lastInProgressDAG
              + ", eventType=" + eventType
              + ", event=" + event.toString());
        }
      }
      dagRecoveryStream.close();
    }

    if (!recoveredDAGData.isCompleted
        && !recoveredDAGData.nonRecoverable) {
      if (lastInProgressDAGData.bufferedSummaryEvents != null
        && !lastInProgressDAGData.bufferedSummaryEvents.isEmpty()) {
        for (HistoryEvent bufferedEvent : lastInProgressDAGData.bufferedSummaryEvents) {
          assert recoveredDAGData.recoveredDAG != null;
          switch (bufferedEvent.getEventType()) {
            case VERTEX_GROUP_COMMIT_STARTED:
              recoveredDAGData.recoveredDAG.restoreFromEvent(bufferedEvent);
              break;
            case VERTEX_GROUP_COMMIT_FINISHED:
              recoveredDAGData.recoveredDAG.restoreFromEvent(bufferedEvent);
              break;
            case VERTEX_FINISHED:
              VertexFinishedEvent vertexFinishedEvent =
                  (VertexFinishedEvent) bufferedEvent;
              Vertex vertex = recoveredDAGData.recoveredDAG.getVertex(
                  vertexFinishedEvent.getVertexID());
              if (vertex == null) {
                recoveredDAGData.nonRecoverable = true;
                recoveredDAGData.reason = "All state could not be recovered"
                    + ", vertex completed but events not flushed"
                    + ", vertexId=" + vertexFinishedEvent.getVertexID();
              } else {
                vertex.restoreFromEvent(vertexFinishedEvent);
              }
              break;
            case DAG_KILL_REQUEST:
              DAGKillRequestEvent killRequestEvent = (DAGKillRequestEvent)bufferedEvent;
              recoveredDAGData.isSessionStopped = killRequestEvent.isSessionStopped();
              recoveredDAGData.recoveredDAG.restoreFromEvent(bufferedEvent);
              break;
            default:
              throw new RuntimeException("Invalid data found in buffered summary events"
                  + ", unknown event type "
                  + bufferedEvent.getEventType());
          }
        }
      }
    }

    return recoveredDAGData;
  }

}