From 8bf87eede2b1735993abc0f2cc49c971f9f8e222 Mon Sep 17 00:00:00 2001
From: Naganarasimha <naganarasimha_gr@apache.org>
Date: Wed, 22 Jun 2016 23:32:19 +0530
Subject: [PATCH] MAPREDUCE-6720. Inconsistent values of counters across tasks
 and job reported to timeline service. Contributed by Varun Saxena

---
 .../mapreduce/jobhistory/JobHistoryEventHandler.java |  4 ++++
 .../mapreduce/jobhistory/JobFinishedEvent.java       | 12 ++++++------
 .../hadoop/mapreduce/util/JobHistoryEventUtils.java  |  7 ++++++-
 3 files changed, 16 insertions(+), 7 deletions(-)

diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
index a09c17b08c..9b59676641 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
+++ b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobHistoryEventHandler.java
@@ -1125,6 +1125,10 @@ private void processEventForTimelineServer(HistoryEvent event, JobId jobId,
     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;
   }
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobFinishedEvent.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobFinishedEvent.java
index 76d689715e..ea21f60341 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/jobhistory/JobFinishedEvent.java
+++ b/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 TimelineEvent toTimelineEvent() {
 
   @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;
   }
 }
diff --git a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/JobHistoryEventUtils.java b/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/JobHistoryEventUtils.java
index 35d066c214..89abf0e30d 100644
--- a/hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-core/src/main/java/org/apache/hadoop/mapreduce/util/JobHistoryEventUtils.java
+++ b/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 static JsonNode countersToJSON(Counters counters) {
 
   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());