svn commit: r1478217 - /incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

[hitesh]

Author: hitesh
Date: Wed May  1 22:13:42 2013
New Revision: 1478217

URL: http://svn.apache.org/r1478217
Log:
TEZ-101. Fix job history event emit for vertex finished to happen only once. 
(hitesh)

Modified:
    
incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java

Modified: 
incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
URL: 
http://svn.apache.org/viewvc/incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java?rev=1478217&r1=1478216&r2=1478217&view=diff
==============================================================================
--- 
incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
 (original)
+++ 
incubator/tez/branches/TEZ-1/tez-dag/src/main/java/org/apache/tez/dag/app/dag/impl/VertexImpl.java
 Wed May  1 22:13:42 2013
@@ -62,6 +62,7 @@ import org.apache.tez.dag.api.EdgeProper
 import org.apache.tez.dag.api.VertexLocationHint;
 import org.apache.tez.dag.api.VertexLocationHint.TaskLocationHint;
 import org.apache.tez.dag.api.client.VertexStatus;
+import org.apache.tez.dag.api.client.VertexStatus.State;
 import org.apache.tez.dag.api.impl.NullVertexOutputCommitter;
 import org.apache.tez.dag.api.impl.VertexContext;
 import org.apache.tez.dag.api.impl.VertexOutputCommitter;
@@ -641,7 +642,6 @@ public class VertexImpl implements org.a
     //check for vertex failure first
     if (vertex.failedTaskCount > 1) {
       vertex.setFinishTime();
-
       String diagnosticMsg = "Vertex failed as tasks failed. "
           + "failedTasks:"
           + vertex.failedTaskCount;
@@ -688,18 +688,22 @@ public class VertexImpl implements org.a
       // metrics.endRunningJob(this);
     }
     if (finishTime == 0) setFinishTime();
-    logJobHistoryVertexFinishedEvent();
 
     switch (finalState) {
       case KILLED:
+      case KILL_WAIT:
         // TODO: Metrics
         //metrics.killedJob(this);
+        logJobHistoryVertexFailedEvent(State.KILLED);
         break;
+      case ERROR:
       case FAILED:
         // TODO: Metrics
         //metrics.failedJob(this);
+        logJobHistoryVertexFailedEvent(State.FAILED);
         break;
       case SUCCEEDED:
+        logJobHistoryVertexFinishedEvent();
         // TODO: Metrics
         //metrics.completedJob(this);
     }
@@ -899,7 +903,6 @@ public class VertexImpl implements org.a
   }
 
   private void abortVertex(VertexStatus.State finalState) {
-    //TODO: Committer?    /*
     try {
       committer.abortVertex(finalState);
     } catch (IOException e) {
@@ -909,7 +912,6 @@ public class VertexImpl implements org.a
     if (finishTime == 0) {
       setFinishTime();
     }
-    logJobHistoryVertexFailedEvent(finalState);
   }
 
   private void mayBeConstructFinalFullCounters() {
@@ -939,7 +941,6 @@ public class VertexImpl implements org.a
     @Override
     public void transition(VertexImpl vertex, VertexEvent event) {
       vertex.setFinishTime();
-      vertex.logJobHistoryVertexFailedEvent(VertexStatus.State.KILLED);
       vertex.finished(VertexState.KILLED);
     }
   }
@@ -1147,8 +1148,6 @@ public class VertexImpl implements org.a
     public void transition(VertexImpl vertex, VertexEvent event) {
       //TODO Is this JH event required.
       vertex.setFinishTime();
-      vertex.logJobHistoryVertexFailedEvent(
-          VertexStatus.State.FAILED);
       vertex.finished(VertexState.ERROR);
     }
   }