MAPREDUCE-6720. Inconsistent values of counters across tasks and job reported to timeline service. Contributed by Varun Saxena

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

@@ -1125,6 +1125,10 @@ public class JobHistoryEventHandler extends AbstractService
     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;
   }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobFinishedEvent.java

@@ -158,12 +158,12 @@ public class JobFinishedEvent  implements HistoryEvent {
 
   @Override
   public Set<TimelineMetric> getTimelineMetrics() {
-    Set<TimelineMetric> jobMetrics = JobHistoryEventUtils
-        .countersToTimelineMetric(getMapCounters(), finishTime);
-    jobMetrics.addAll(JobHistoryEventUtils
-        .countersToTimelineMetric(getReduceCounters(), finishTime));
-    jobMetrics.addAll(JobHistoryEventUtils
-        .countersToTimelineMetric(getTotalCounters(), finishTime));
+    Set<TimelineMetric> jobMetrics = JobHistoryEventUtils.
+        countersToTimelineMetric(getTotalCounters(), finishTime);
+    jobMetrics.addAll(JobHistoryEventUtils.
+        countersToTimelineMetric(getMapCounters(), finishTime, "MAP:"));
+    jobMetrics.addAll(JobHistoryEventUtils.
+        countersToTimelineMetric(getReduceCounters(), finishTime, "REDUCE:"));
     return jobMetrics;
   }
 }

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/JobHistoryEventUtils.java

@@ -61,11 +61,16 @@ public final class JobHistoryEventUtils {
 
   public static Set<TimelineMetric> countersToTimelineMetric(Counters counters,
       long timestamp) {
+    return countersToTimelineMetric(counters, timestamp, "");
+  }
+
+  public static Set<TimelineMetric> countersToTimelineMetric(Counters counters,
+      long timestamp, String groupNamePrefix) {
     Set<TimelineMetric> entityMetrics = new HashSet<TimelineMetric>();
     for (CounterGroup g : counters) {
       String groupName = g.getName();
       for (Counter c : g) {
-        String name = groupName + ":" + c.getName();
+        String name = groupNamePrefix + groupName + ":" + c.getName();
         TimelineMetric metric = new TimelineMetric();
         metric.setId(name);
         metric.addValue(timestamp, c.getValue());