svn commit: r647144 - in /hadoop/core/trunk: CHANGES.txt src/java/org/apache/hadoop/mapred/ReduceTask.java

[dd...@apache.org]

Author: ddas
Date: Fri Apr 11 05:39:19 2008
New Revision: 647144

URL: http://svn.apache.org/viewvc?rev=647144&view=rev
Log:
HADOOP-3204. Fixes a problem to do with ReduceTask's LocalFSMerger not catching Throwable. Contributed by Amar Ramesh Kamat.

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

Modified: hadoop/core/trunk/CHANGES.txt
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/CHANGES.txt?rev=647144&r1=647143&r2=647144&view=diff
==============================================================================
--- hadoop/core/trunk/CHANGES.txt (original)
+++ hadoop/core/trunk/CHANGES.txt Fri Apr 11 05:39:19 2008
@@ -568,6 +568,9 @@
    HADOOP-1373. checkPath() should ignore case when it compares authoriy.
    (Edward J. Yoon via rangadi)
 
+   HADOOP-3204. Fixes a problem to do with ReduceTask's LocalFSMerger not
+   catching Throwable.  (Amar Ramesh Kamat via ddas)
+
 Release 0.16.3 - Unreleased
 
   BUG FIXES

Modified: hadoop/core/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java
URL: http://svn.apache.org/viewvc/hadoop/core/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java?rev=647144&r1=647143&r2=647144&view=diff
==============================================================================
--- hadoop/core/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java (original)
+++ hadoop/core/trunk/src/java/org/apache/hadoop/mapred/ReduceTask.java Fri Apr 11 05:39:19 2008
@@ -1380,6 +1380,10 @@
             LOG.warn(reduceTask.getTaskId() +
                      " Final merge of the inmemory files threw an exception: " + 
                      StringUtils.stringifyException(t));
+            // check if the last merge generated an error
+            if (mergeThrowable != null) {
+              mergeThrowable = t;
+            }
             return false;
           }
         }
@@ -1548,9 +1552,15 @@
             .suffix(".merged");
           SequenceFile.Writer writer =
             sorter.cloneFileAttributes(mapFiles[0], outputPath, null);
-          SequenceFile.Sorter.RawKeyValueIterator iter;
+          SequenceFile.Sorter.RawKeyValueIterator iter  = null;
           Path tmpDir = new Path(reduceTask.getTaskId());
-          iter = sorter.merge(mapFiles, true, ioSortFactor, tmpDir);
+          try {
+            iter = sorter.merge(mapFiles, true, ioSortFactor, tmpDir);
+          } catch (Exception e) {
+            writer.close();
+            localFileSys.delete(outputPath, true);
+            throw new IOException (StringUtils.stringifyException(e));
+          }
           sorter.writeFile(iter, writer);
           writer.close();
           
@@ -1560,12 +1570,12 @@
           
           LOG.info(reduceTask.getTaskId()
                    + " Finished merging map output files on disk.");
-        } catch (IOException ioe) {
+        } catch (Throwable t) {
           LOG.warn(reduceTask.getTaskId()
                    + " Merging of the local FS files threw an exception: "
-                   + StringUtils.stringifyException(ioe));
+                   + StringUtils.stringifyException(t));
           if (mergeThrowable == null) {
-            mergeThrowable = ioe;
+            mergeThrowable = t;
           }
         } finally {
           localFSMergeInProgress = false;