MAPREDUCE-6892. Issues with the count of failed/killed tasks in the j…

…hist file. (Peter Bacsko via Haibo Chen)

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java

@@ -431,10 +431,18 @@ protected void serviceStop() throws Exception {
             + " 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);
+          int successfulMaps = job.getCompletedMaps() - job.getFailedMaps()
+                  - job.getKilledMaps();
+          int successfulReduces = job.getCompletedReduces()
+                  - job.getFailedReduces() - job.getKilledReduces();
+
           JobUnsuccessfulCompletionEvent jucEvent =
             new JobUnsuccessfulCompletionEvent(TypeConverter.fromYarn(toClose),
-                System.currentTimeMillis(), job.getCompletedMaps(),
-                job.getCompletedReduces(),
+                System.currentTimeMillis(),
+                successfulMaps,
+                successfulReduces,
+                job.getFailedMaps(), job.getFailedReduces(),
+                job.getKilledMaps(), job.getKilledReduces(),
                 createJobStateForJobUnsuccessfulCompletionEvent(
                     mi.getForcedJobStateOnShutDown()),
                 job.getDiagnostics());
@@ -655,9 +663,9 @@ public void handleEvent(JobHistoryEvent event) {
           JobFinishedEvent jFinishedEvent =
               (JobFinishedEvent) event.getHistoryEvent();
           mi.getJobIndexInfo().setFinishTime(jFinishedEvent.getFinishTime());
-          mi.getJobIndexInfo().setNumMaps(jFinishedEvent.getFinishedMaps());
+          mi.getJobIndexInfo().setNumMaps(jFinishedEvent.getSucceededMaps());
           mi.getJobIndexInfo().setNumReduces(
-              jFinishedEvent.getFinishedReduces());
+              jFinishedEvent.getSucceededReduces());
           mi.getJobIndexInfo().setJobStatus(JobState.SUCCEEDED.toString());
           closeEventWriter(event.getJobID());
           processDoneFiles(event.getJobID());
@@ -672,8 +680,8 @@ public void handleEvent(JobHistoryEvent event) {
           JobUnsuccessfulCompletionEvent jucEvent =
               (JobUnsuccessfulCompletionEvent) event.getHistoryEvent();
           mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
-          mi.getJobIndexInfo().setNumMaps(jucEvent.getFinishedMaps());
-          mi.getJobIndexInfo().setNumReduces(jucEvent.getFinishedReduces());
+          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
+          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
           mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
           closeEventWriter(event.getJobID());
           if(context.isLastAMRetry())
@@ -690,8 +698,8 @@ public void handleEvent(JobHistoryEvent event) {
               (JobUnsuccessfulCompletionEvent) event
               .getHistoryEvent();
           mi.getJobIndexInfo().setFinishTime(jucEvent.getFinishTime());
-          mi.getJobIndexInfo().setNumMaps(jucEvent.getFinishedMaps());
-          mi.getJobIndexInfo().setNumReduces(jucEvent.getFinishedReduces());
+          mi.getJobIndexInfo().setNumMaps(jucEvent.getSucceededMaps());
+          mi.getJobIndexInfo().setNumReduces(jucEvent.getSucceededReduces());
           mi.getJobIndexInfo().setJobStatus(jucEvent.getStatus());
           closeEventWriter(event.getJobID());
           processDoneFiles(event.getJobID());
@@ -739,10 +747,12 @@ public void processEventForJobSummary(HistoryEvent event, JobSummary summary,
     case JOB_FINISHED:
       JobFinishedEvent jfe = (JobFinishedEvent) event;
       summary.setJobFinishTime(jfe.getFinishTime());
-      summary.setNumFinishedMaps(jfe.getFinishedMaps());
+      summary.setNumSucceededMaps(jfe.getSucceededMaps());
       summary.setNumFailedMaps(jfe.getFailedMaps());
-      summary.setNumFinishedReduces(jfe.getFinishedReduces());
+      summary.setNumSucceededReduces(jfe.getSucceededReduces());
       summary.setNumFailedReduces(jfe.getFailedReduces());
+      summary.setNumKilledMaps(jfe.getKilledMaps());
+      summary.setNumKilledReduces(jfe.getKilledReduces());
       if (summary.getJobStatus() == null)
         summary
             .setJobStatus(org.apache.hadoop.mapreduce.JobStatus.State.SUCCEEDED
@@ -753,11 +763,21 @@ public void processEventForJobSummary(HistoryEvent event, JobSummary summary,
       break;
     case JOB_FAILED:
     case JOB_KILLED:
+      Job job = context.getJob(jobId);
       JobUnsuccessfulCompletionEvent juce = (JobUnsuccessfulCompletionEvent) event;
+      int successfulMaps = job.getCompletedMaps() - job.getFailedMaps()
+          - job.getKilledMaps();
+      int successfulReduces = job.getCompletedReduces()
+          - job.getFailedReduces() - job.getKilledReduces();
+
       summary.setJobStatus(juce.getStatus());
-      summary.setNumFinishedMaps(context.getJob(jobId).getTotalMaps());
-      summary.setNumFinishedReduces(context.getJob(jobId).getTotalReduces());
+      summary.setNumSucceededMaps(successfulMaps);
+      summary.setNumSucceededReduces(successfulReduces);
+      summary.setNumFailedMaps(job.getFailedMaps());
+      summary.setNumFailedReduces(job.getFailedReduces());
       summary.setJobFinishTime(juce.getFinishTime());
+      summary.setNumKilledMaps(juce.getKilledMaps());
+      summary.setNumKilledReduces(juce.getKilledReduces());
       setSummarySlotSeconds(summary, context.getJob(jobId).getAllCounters());
       break;
     default:
@@ -840,25 +860,43 @@ private void processEventForTimelineServer(HistoryEvent event, JobId jobId,
         JobUnsuccessfulCompletionEvent juce =
               (JobUnsuccessfulCompletionEvent) event;
         tEvent.addEventInfo("FINISH_TIME", juce.getFinishTime());
-        tEvent.addEventInfo("NUM_MAPS", juce.getFinishedMaps());
-        tEvent.addEventInfo("NUM_REDUCES", juce.getFinishedReduces());
+        tEvent.addEventInfo("NUM_MAPS",
+            juce.getSucceededMaps() +
+            juce.getFailedMaps() +
+            juce.getKilledMaps());
+        tEvent.addEventInfo("NUM_REDUCES",
+            juce.getSucceededReduces() +
+            juce.getFailedReduces() +
+            juce.getKilledReduces());
         tEvent.addEventInfo("JOB_STATUS", juce.getStatus());
         tEvent.addEventInfo("DIAGNOSTICS", juce.getDiagnostics());
-        tEvent.addEventInfo("FINISHED_MAPS", juce.getFinishedMaps());
-        tEvent.addEventInfo("FINISHED_REDUCES", juce.getFinishedReduces());
+        tEvent.addEventInfo("SUCCESSFUL_MAPS", juce.getSucceededMaps());
+        tEvent.addEventInfo("SUCCESSFUL_REDUCES", juce.getSucceededReduces());
+        tEvent.addEventInfo("FAILED_MAPS", juce.getFailedMaps());
+        tEvent.addEventInfo("FAILED_REDUCES", juce.getFailedReduces());
+        tEvent.addEventInfo("KILLED_MAPS", juce.getKilledMaps());
+        tEvent.addEventInfo("KILLED_REDUCES", juce.getKilledReduces());
         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("NUM_MAPS",
+            jfe.getSucceededMaps() +
+            jfe.getFailedMaps() +
+            jfe.getKilledMaps());
+        tEvent.addEventInfo("NUM_REDUCES",
+            jfe.getSucceededReduces() +
+            jfe.getFailedReduces() +
+            jfe.getKilledReduces());
         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("SUCCESSFUL_MAPS", jfe.getSucceededMaps());
+        tEvent.addEventInfo("SUCCESSFUL_REDUCES", jfe.getSucceededReduces());
+        tEvent.addEventInfo("KILLED_MAPS", jfe.getKilledMaps());
+        tEvent.addEventInfo("KILLED_REDUCES", jfe.getKilledReduces());
         tEvent.addEventInfo("MAP_COUNTERS_GROUPS",
             JobHistoryEventUtils.countersToJSON(jfe.getMapCounters()));
         tEvent.addEventInfo("REDUCE_COUNTERS_GROUPS",

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobSummary.java

@@ -30,10 +30,12 @@ public class JobSummary {
   private long firstReduceTaskLaunchTime; // ReduceAttemptStarted |
                                           // TaskAttemptStartEvent
   private long jobFinishTime;
-  private int numFinishedMaps;
+  private int numSucceededMaps;
   private int numFailedMaps;
-  private int numFinishedReduces;
+  private int numSucceededReduces;
   private int numFailedReduces;
+  private int numKilledMaps;
+  private int numKilledReduces;
   private int resourcesPerMap; // resources used per map/min resource
   private int resourcesPerReduce; // resources used per reduce/min resource
   // resource models
@@ -98,12 +100,12 @@ public void setJobFinishTime(long jobFinishTime) {
     this.jobFinishTime = jobFinishTime;
   }
 
-  public int getNumFinishedMaps() {
-    return numFinishedMaps;
+  public int getNumSucceededMaps() {
+    return numSucceededMaps;
   }
 
-  public void setNumFinishedMaps(int numFinishedMaps) {
-    this.numFinishedMaps = numFinishedMaps;
+  public void setNumSucceededMaps(int numSucceededMaps) {
+    this.numSucceededMaps = numSucceededMaps;
   }
 
   public int getNumFailedMaps() {
@@ -114,6 +116,22 @@ public void setNumFailedMaps(int numFailedMaps) {
     this.numFailedMaps = numFailedMaps;
   }
 
+  public int getKilledMaps() {
+    return numKilledMaps;
+  }
+
+  public void setNumKilledMaps(int numKilledMaps) {
+    this.numKilledMaps = numKilledMaps;
+  }
+
+  public int getKilledReduces() {
+    return numKilledReduces;
+  }
+
+  public void setNumKilledReduces(int numKilledReduces) {
+    this.numKilledReduces = numKilledReduces;
+  }
+
   public int getResourcesPerMap() {
     return resourcesPerMap;
   }
@@ -122,12 +140,12 @@ public void setResourcesPerMap(int resourcesPerMap) {
     this.resourcesPerMap = resourcesPerMap;
   }
 
-  public int getNumFinishedReduces() {
-    return numFinishedReduces;
+  public int getNumSucceededReduces() {
+    return numSucceededReduces;
   }
 
-  public void setNumFinishedReduces(int numFinishedReduces) {
-    this.numFinishedReduces = numFinishedReduces;
+  public void setNumSucceededReduces(int numSucceededReduces) {
+    this.numSucceededReduces = numSucceededReduces;
   }
 
   public int getNumFailedReduces() {
@@ -204,8 +222,15 @@ public String getJobSummaryString() {
       .add("finishTime", jobFinishTime)
       .add("resourcesPerMap", resourcesPerMap)
       .add("resourcesPerReduce", resourcesPerReduce)
-      .add("numMaps", numFinishedMaps + numFailedMaps)
-      .add("numReduces", numFinishedReduces + numFailedReduces)
+      .add("numMaps", numSucceededMaps + numFailedMaps + numKilledMaps)
+      .add("numReduces", numSucceededReduces + numFailedReduces
+          + numKilledReduces)
+      .add("succededMaps", numSucceededMaps)
+      .add("succeededReduces", numSucceededReduces)
+      .add("failedMaps", numFailedMaps)
+      .add("failedReduces", numFailedReduces)
+      .add("killedMaps", numKilledMaps)
+      .add("killedReduces", numKilledReduces)
       .add("user", user)
       .add("queue", queue)
       .add("status", jobStatus)

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/Job.java

@@ -65,6 +65,10 @@ public interface Job {
   int getTotalReduces();
   int getCompletedMaps();
   int getCompletedReduces();
+  int getFailedMaps();
+  int getFailedReduces();
+  int getKilledMaps();
+  int getKilledReduces();
   float getProgress();
   boolean isUber();
   String getUserName();

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/JobImpl.java

@@ -1684,6 +1684,10 @@ private void unsuccessfulFinish(JobStateInternal finalState) {
               finishTime,
               succeededMapTaskCount,
               succeededReduceTaskCount,
+              failedMapTaskCount,
+              failedReduceTaskCount,
+              killedMapTaskCount,
+              killedReduceTaskCount,
               finalState.toString(),
               diagnostics);
       eventHandler.handle(new JobHistoryEvent(jobId,
@@ -1748,6 +1752,7 @@ private static JobFinishedEvent createJobFinishedEvent(JobImpl job) {
         job.oldJobId, job.finishTime,
         job.succeededMapTaskCount, job.succeededReduceTaskCount,
         job.failedMapTaskCount, job.failedReduceTaskCount,
+        job.killedMapTaskCount, job.killedReduceTaskCount,
         job.finalMapCounters,
         job.finalReduceCounters,
         job.fullCounters);
@@ -1797,7 +1802,7 @@ public void transition(JobImpl job, JobEvent event) {
       job.setFinishTime();
       JobUnsuccessfulCompletionEvent failedEvent =
           new JobUnsuccessfulCompletionEvent(job.oldJobId,
-              job.finishTime, 0, 0,
+              job.finishTime, 0, 0, 0, 0, 0, 0,
               JobStateInternal.KILLED.toString(), job.diagnostics);
       job.eventHandler.handle(new JobHistoryEvent(job.jobId, failedEvent));
       job.finished(JobStateInternal.KILLED);
@@ -1954,8 +1959,8 @@ private static class TaskCompletedTransition implements
     @Override
     public JobStateInternal transition(JobImpl job, JobEvent event) {
       job.completedTaskCount++;
-      LOG.info("Num completed Tasks: " + job.completedTaskCount);
       JobTaskEvent taskEvent = (JobTaskEvent) event;
+      LOG.info("Num completed Tasks: " + job.completedTaskCount);
       Task task = job.tasks.get(taskEvent.getTaskID());
       if (taskEvent.getState() == TaskState.SUCCEEDED) {
         taskSucceeded(job, task);
@@ -1991,11 +1996,15 @@ protected JobStateInternal checkJobAfterTaskCompletion(JobImpl job) {
         job.allowedMapFailuresPercent*job.numMapTasks ||
         job.failedReduceTaskCount*100 > 
         job.allowedReduceFailuresPercent*job.numReduceTasks) {
+
         job.setFinishTime();
 
         String diagnosticMsg = "Job failed as tasks failed. " +
             "failedMaps:" + job.failedMapTaskCount + 
-            " failedReduces:" + job.failedReduceTaskCount;
+            " failedReduces:" + job.failedReduceTaskCount +
+            " killedMaps:" + job.killedMapTaskCount +
+            " killedReduces: " + job.killedReduceTaskCount;
+
         LOG.info(diagnosticMsg);
         job.addDiagnostic(diagnosticMsg);
 
@@ -2226,7 +2235,13 @@ public void transition(JobImpl job, JobEvent event) {
       job.setFinishTime();
       JobUnsuccessfulCompletionEvent failedEvent =
           new JobUnsuccessfulCompletionEvent(job.oldJobId,
-              job.finishTime, 0, 0,
+              job.finishTime,
+              job.succeededMapTaskCount,
+              job.succeededReduceTaskCount,
+              job.failedMapTaskCount,
+              job.failedReduceTaskCount,
+              job.killedMapTaskCount,
+              job.killedReduceTaskCount,
               jobHistoryString, job.diagnostics);
       job.eventHandler.handle(new JobHistoryEvent(job.jobId, failedEvent));
       job.finished(terminationState);
@@ -2266,4 +2281,24 @@ public int getMaxFetchFailuresNotifications() {
   public void setJobPriority(Priority priority) {
     this.jobPriority = priority;
   }
+
+  @Override
+  public int getFailedMaps() {
+    return failedMapTaskCount;
+  }
+
+  @Override
+  public int getFailedReduces() {
+    return failedReduceTaskCount;
+  }
+
+  @Override
+  public int getKilledMaps() {
+    return killedMapTaskCount;
+  }
+
+  @Override
+  public int getKilledReduces() {
+    return killedReduceTaskCount;
+  }
 }