svn commit: r807064 - in /hadoop/mapreduce/trunk: CHANGES.txt src/java/org/apache/hadoop/mapred/ReduceTask.java src/java/org/apache/hadoop/mapred/TaskTracker.java

[cd...@apache.org]

Author: cdouglas
Date: Mon Aug 24 03:53:21 2009
New Revision: 807064

URL: http://svn.apache.org/viewvc?rev=807064&view=rev
Log:
MAPREDUCE-479. Provide full task id to map output servlet rather than the
reduce id, only. Contributed by Jiaqi Tan


Modified:
    hadoop/mapreduce/trunk/CHANGES.txt
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java
    hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java

Modified: hadoop/mapreduce/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/CHANGES.txt?rev=807064&r1=807063&r2=807064&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/CHANGES.txt (original)
+++ hadoop/mapreduce/trunk/CHANGES.txt Mon Aug 24 03:53:21 2009
@@ -27,6 +27,9 @@
     FileNotFoundException from FileSystem::listStatus rather than returning
     null. (Jakob Homan via cdouglas)
 
+    MAPREDUCE-479. Provide full task id to map output servlet rather than the
+    reduce id, only. (Jiaqi Tan via cdouglas)
+
   NEW FEATURES
 
     MAPREDUCE-706. Support for FIFO pools in the fair scheduler.

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java?rev=807064&r1=807063&r2=807064&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java Mon Aug 24 03:53:21 2009
@@ -2760,10 +2760,12 @@
                 maxFetchRetriesPerMap = Math.max(MIN_FETCH_RETRIES_PER_MAP, 
                   getClosestPowerOf2((maxMapRuntime / BACKOFF_INIT) + 1));
               }
+              // send the full attempt ID of the reduce task to enable full 
+              // causal tracing of shuffle from map attempt to reduce attempt
               URL mapOutputLocation = new URL(event.getTaskTrackerHttp() + 
                                       "/mapOutput?job=" + taskId.getJobID() +
                                       "&map=" + taskId + 
-                                      "&reduce=" + getPartition());
+                                      "&reduce=" + reduceTask.getTaskID());
               List<MapOutputLocation> loc = mapLocations.get(host);
               if (loc == null) {
                 loc = Collections.synchronizedList

Modified: hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java
URL: http://svn.apache.org/viewvc/hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java?rev=807064&r1=807063&r2=807064&view=diff
==============================================================================
--- hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java (original)
+++ hadoop/mapreduce/trunk/src/java/org/apache/hadoop/mapred/TaskTracker.java Mon Aug 24 03:53:21 2009
@@ -131,6 +131,7 @@
         ", bytes: %s" + // byte count
         ", op: %s" +    // operation
         ", cliID: %s" + // task id
+        ", reduceID: %s" + // reduce id
         ", duration: %s"; // duration
   public static final Log ClientTraceLog =
     LogFactory.getLog(TaskTracker.class.getName() + ".clienttrace");
@@ -3160,6 +3161,7 @@
     public void doGet(HttpServletRequest request, 
                       HttpServletResponse response
                       ) throws ServletException, IOException {
+      TaskAttemptID reduceAttId = null;
       String mapId = request.getParameter("map");
       String reduceId = request.getParameter("reduce");
       String jobId = request.getParameter("job");
@@ -3171,8 +3173,13 @@
       if (mapId == null || reduceId == null) {
         throw new IOException("map and reduce parameters are required");
       }
+      try {
+        reduceAttId = TaskAttemptID.forName(reduceId);
+      } catch (IllegalArgumentException e) {
+        throw new IOException("reduce attempt ID is malformed");
+      }
       ServletContext context = getServletContext();
-      int reduce = Integer.parseInt(reduceId);
+      int reduce = reduceAttId.getTaskID().getId();
       byte[] buffer = new byte[MAX_BYTES_TO_READ];
       // true iff IOException was caused by attempt to access input
       boolean isInputException = true;
@@ -3287,7 +3294,8 @@
           ClientTraceLog.info(String.format(MR_CLIENTTRACE_FORMAT,
                 request.getLocalAddr() + ":" + request.getLocalPort(),
                 request.getRemoteAddr() + ":" + request.getRemotePort(),
-                totalRead, "MAPRED_SHUFFLE", mapId, endTime-startTime));
+                totalRead, "MAPRED_SHUFFLE", mapId, reduceId,
+                endTime-startTime));
         }
       }
       outStream.close();