HADOOP-597. Fix TaskTracker to not discard map outputs for errors in transmission to reduce node. Contributed by Owen.

git-svn-id: https://svn.apache.org/repos/asf/lucene/hadoop/trunk@462918 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -16,6 +16,9 @@ Release 0.7.1 - unreleased
  4. HADOOP-598.  Fix tasks to retry when reporting completion, so that
     a single RPC timeout won't fail a task.  (omalley via cutting)
 
+ 5. HADOOP-597.  Fix TaskTracker to not discard map outputs for errors
+    in transmitting them to reduce nodes.  (omalley via cutting)
+
 
 Release 0.7.0 - 2006-10-06
 

src/java/org/apache/hadoop/mapred/TaskTracker.java

@@ -1365,30 +1365,39 @@ public class TaskTracker
         Path filename = conf.getLocalPath(mapId+"/part-"+reduce+".out");
         response.setContentLength((int) fileSys.getLength(filename));
         InputStream inStream = null;
+        // true iff IOException was caused by attempt to access input
+        boolean isInputException = true;
         try {
           inStream = fileSys.open(filename);
           try {
             int len = inStream.read(buffer);
             while (len > 0) {
-              outStream.write(buffer, 0, len);
+              try {
+                outStream.write(buffer, 0, len);
+              } catch (IOException ie) {
+                isInputException = false;
+                throw ie;
+              }
               len = inStream.read(buffer);
             }
           } finally {
             inStream.close();
-            outStream.close();
           }
         } catch (IOException ie) {
           TaskTracker tracker = 
             (TaskTracker) context.getAttribute("task.tracker");
           Log log = (Log) context.getAttribute("log");
-          String errorMsg = "getMapOutput(" + mapId + "," + reduceId + 
-          ") failed :\n"+
-          StringUtils.stringifyException(ie);
+          String errorMsg = ("getMapOutput(" + mapId + "," + reduceId + 
+                             ") failed :\n"+
+                             StringUtils.stringifyException(ie));
           log.warn(errorMsg);
-          tracker.mapOutputLost(mapId, errorMsg);
+          if (isInputException) {
+            tracker.mapOutputLost(mapId, errorMsg);
+          }
           response.sendError(HttpServletResponse.SC_GONE, errorMsg);
           throw ie;
         } 
+        outStream.close();
       }
     }
 }