MAPREDUCE-2661. Fix TaskImpl to not access MapTaskImpl. Contributed by Ahmed Radwan.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/MR-279@1148209 13f79535-47bb-0310-9956-ffa450edef68

mapreduce/CHANGES.txt

@@ -4,7 +4,10 @@ Trunk (unreleased changes)
 
 
     MAPREDUCE-279
-   
+    
+    MAPREDUCE-2661. Fix TaskImpl to not access MapTaskImpl. (Ahmed Radwan 
+    via sharad)  
+ 
     MAPREDUCE-2644. NodeManager fails to create containers when NM_LOG_DIR
     is not explicitly set in the Configuration. (Josh Wills via vinodkv)
 

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/MapTaskImpl.java

@@ -78,4 +78,20 @@ public class MapTaskImpl extends TaskImpl {
   protected TaskSplitMetaInfo getTaskSplitMetaInfo() {
     return this.taskSplitMetaInfo;
   }
+
+  /**
+   * @return a String formatted as a comma-separated list of splits.
+   */
+  @Override
+  protected String getSplitsAsString() {
+    String[] splits = getTaskSplitMetaInfo().getLocations();
+    if (splits == null || splits.length == 0)
+    return "";
+    StringBuilder sb = new StringBuilder();
+    for (int i = 0; i < splits.length; i++) {
+      if (i != 0) sb.append(",");
+      sb.append(splits[i]);
+    }
+    return sb.toString();
+  }
 }

mapreduce/mr-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/job/impl/TaskImpl.java

@@ -601,7 +601,18 @@ public abstract class TaskImpl implements Task, EventHandler<TaskEvent> {
         taId == null ? null : TypeConverter.fromYarn(taId));
     return taskFailedEvent;
   }
-  
+
+  /**
+  * @return a String representation of the splits.
+  *
+  * Subclasses can override this method to provide their own representations
+  * of splits (if any).
+  *
+  */
+  protected String getSplitsAsString(){
+	  return "";
+  }
+
   private static class InitialScheduleTransition
     implements SingleArcTransition<TaskImpl, TaskEvent> {
 
@@ -612,23 +623,11 @@ public abstract class TaskImpl implements Task, EventHandler<TaskEvent> {
       TaskStartedEvent tse = new TaskStartedEvent(
           TypeConverter.fromYarn(task.taskId), task.getLaunchTime(),
           TypeConverter.fromYarn(task.taskId.getTaskType()),
-          task instanceof MapTaskImpl ? splitsAsString(((MapTaskImpl) task) //TODO Should not be accessing MapTaskImpl
-              .getTaskSplitMetaInfo().getLocations()) : "");
+          task.getSplitsAsString());
       task.eventHandler
           .handle(new JobHistoryEvent(task.taskId.getJobId(), tse));
       task.historyTaskStartGenerated = true;
     }
-    
-    private String splitsAsString(String[] splits) {
-      if (splits == null || splits.length == 0)
-        return "";
-      StringBuilder sb = new StringBuilder();
-      for (int i = 0; i < splits.length; i++) {
-        if (i != 0) sb.append(",");
-        sb.append(splits[i]);
-      }
-      return sb.toString();
-    }
   }
 
   // Used when creating a new attempt while one is already running.