HADOOP-4620. Fixes Streaming to handle well the cases of map/reduce with empty input/output. Contributed by Ravi Gummadi.

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

CHANGES.txt

@@ -1446,6 +1446,9 @@ Release 0.18.3 - Unreleased
     HADOOP-4795. Prevent lease monitor getting into an infinite loop when
     leases and the namespace tree does not match. (szetszwo)
 
+    HADOOP-4620. Fixes Streaming to handle well the cases of map/reduce with empty
+    input/output. (Ravi Gummadi via ddas)
+
 Release 0.18.2 - 2008-11-03
 
   BUG FIXES

src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapRed.java

@@ -288,6 +288,22 @@ public abstract class PipeMapRed {
 
   void waitOutputThreads() {
     try {
+      if (outThread_ == null) {
+        // This happens only when reducer has empty input(So reduce() is not
+        // called at all in this task). If reducer still generates output,
+        // which is very uncommon and we may not have to support this case.
+        // So we don't write this output to HDFS, but we consume/collect
+        // this output just to avoid reducer hanging forever.
+
+        OutputCollector collector = new OutputCollector() {
+          public void collect(Object key, Object value)
+            throws IOException {
+            //just consume it, no need to write the record anywhere
+          }
+        };
+        Reporter reporter = Reporter.NULL;//dummy reporter
+        startOutputThreads(collector, reporter);
+      }
       int exitVal = sim.waitFor();
       // how'd it go?
       if (exitVal != 0) {
@@ -506,9 +522,11 @@ public abstract class PipeMapRed {
   }
 
   public void mapRedFinished() {
-    logprintln("mapRedFinished");
     try {
-      if (!doPipe_) return;
+      if (!doPipe_) {
+        logprintln("mapRedFinished");
+        return;
+      }
       try {
         if (clientOut_ != null) {
           clientOut_.flush();
@@ -518,6 +536,7 @@ public abstract class PipeMapRed {
       }
       waitOutputThreads();
       if (sim != null) sim.destroy();
+      logprintln("mapRedFinished");
     } catch (RuntimeException e) {
       logStackTrace(e);
       throw e;

src/contrib/streaming/src/java/org/apache/hadoop/streaming/PipeMapper.java

@@ -82,10 +82,6 @@ public class PipeMapper extends PipeMapRed implements Mapper {
   // (MapRed creates it reflectively)
 
   public void map(Object key, Object value, OutputCollector output, Reporter reporter) throws IOException {
-    // init
-    if (outThread_ == null) {
-      startOutputThreads(output, reporter);
-    }
     if (outerrThreadsThrowable != null) {
       mapRedFinished();
       throw new IOException ("MROutput/MRErrThread failed:"

src/contrib/streaming/src/java/org/apache/hadoop/streaming/StreamJob.java

@@ -750,6 +750,7 @@ public class StreamJob implements Tool {
         jobConf_.setMapperClass(c);
       } else {
         jobConf_.setMapperClass(PipeMapper.class);
+        jobConf_.setMapRunnerClass(PipeMapRunner.class);
         jobConf_.set("stream.map.streamprocessor", 
                      URLEncoder.encode(mapCmd_, "UTF-8"));
       }

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

@@ -58,4 +58,7 @@ public class MapRunner<K1, V1, K2, V2>
     }
   }
 
+  protected Mapper<K1, V1, K2, V2> getMapper() {
+    return mapper;
+  }
 }