} catch (IOException e) {
LOG.warn("Failed to persist recovery event for DAG completion"
+ ", dagId=" + dag.dagId
+ ", finalState=" + endState);
}
dag.eventHandler.handle(new DAGAppMasterEventDAGFinished(dag.getID(),
endState));
return endState;
}
for (Vertex v : dag.vertices.values()) {
if (v.getInputVerticesCount() == 0) {
if (LOG.isDebugEnabled()) {
LOG.debug("Sending Running Recovery event to root vertex "
+ v.getName());
}
dag.eventHandler.handle(new VertexEventRecoverVertex(v.getVertexId(),
VertexState.RUNNING));
}
}
return DAGState.RUNNING;
case SUCCEEDED:
case ERROR:
case FAILED:
case KILLED:
// Completed
// Recover all other data for all vertices
// send recover event to all vertices with a final end state
for (Vertex v : dag.vertices.values()) {
VertexState desiredState = VertexState.SUCCEEDED;
if (dag.recoveredState.equals(DAGState.KILLED)) {
desiredState = VertexState.KILLED;
} else if (EnumSet.of(DAGState.ERROR, DAGState.FAILED).contains(
dag.recoveredState)) {
desiredState = VertexState.FAILED;
}
dag.eventHandler.handle(new VertexEventRecoverVertex(v.getVertexId(),
desiredState));
}
// Let us inform AM of completion
dag.eventHandler.handle(new DAGAppMasterEventDAGFinished(dag.getID(),
dag.recoveredState));
LOG.info("Recovered DAG: " + dag.getID() + " finished with state: "
+ dag.recoveredState);
return dag.recoveredState;
default:
// Error state
LOG.warn("Trying to recover DAG, failed to recover"
+ " from non-handled state" + dag.recoveredState);
// Tell AM ERROR so that it can shutdown
dag.eventHandler.handle(new DAGAppMasterEventDAGFinished(dag.getID(),
DAGState.ERROR));
return DAGState.FAILED;
}
}