MAPREDUCE-5113. Streaming input/output types are ignored with java mapper/reducer. (sandyr via tucu)

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

hadoop-mapreduce-project/CHANGES.txt

@@ -158,6 +158,21 @@ Trunk (Unreleased)
     MAPREDUCE-5006. Fix failing streaming tests due to MAPREDUCE-4994.
     (Sandy Ryza via tomwhite)
 
+Release 2.0.5-alpha - UNRELEASED
+
+  INCOMPATIBLE CHANGES
+
+  NEW FEATURES
+
+  IMPROVEMENTS
+
+  OPTIMIZATIONS
+
+  BUG FIXES
+
+    MAPREDUCE-5113. Streaming input/output types are ignored with java 
+    mapper/reducer. (sandyr via tucu)
+
 Release 2.0.4-beta - UNRELEASED
 
   INCOMPATIBLE CHANGES

hadoop-tools/hadoop-streaming/src/main/java/org/apache/hadoop/streaming/StreamJob.java

@@ -875,7 +875,7 @@ public class StreamJob implements Tool {
         IdentifierResolver.TEXT_ID));
     jobConf_.setClass("stream.map.output.reader.class",
       idResolver.getOutputReaderClass(), OutputReader.class);
-    if (isMapperACommand) {
+    if (isMapperACommand || jobConf_.get("stream.map.output") != null) {
       // if mapper is a command, then map output key/value classes come from the
       // idResolver
       jobConf_.setMapOutputKeyClass(idResolver.getOutputKeyClass());
@@ -891,7 +891,7 @@ public class StreamJob implements Tool {
         IdentifierResolver.TEXT_ID));
     jobConf_.setClass("stream.reduce.output.reader.class",
       idResolver.getOutputReaderClass(), OutputReader.class);
-    if (isReducerACommand) {
+    if (isReducerACommand || jobConf_.get("stream.reduce.output") != null) {
       // if reducer is a command, then output key/value classes come from the
       // idResolver
       jobConf_.setOutputKeyClass(idResolver.getOutputKeyClass());

hadoop-tools/hadoop-streaming/src/test/java/org/apache/hadoop/streaming/TestStreamingOutputKeyValueTypes.java

@@ -24,6 +24,7 @@ import org.apache.hadoop.mapred.MapReduceBase;
 import org.apache.hadoop.mapred.OutputCollector;
 import org.apache.hadoop.mapred.Reducer;
 import org.apache.hadoop.mapred.Reporter;
+import org.apache.hadoop.mapred.TextInputFormat;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.junit.Before;
 import org.junit.Test;
@@ -177,6 +178,17 @@ public class TestStreamingOutputKeyValueTypes extends TestStreaming {
     args.add("0");
     super.testCommandLine();
   }
+  
+  @Test
+  public void testDefaultToIdentityReducer() throws Exception {
+    args.add("-mapper");args.add(map);
+    args.add("-jobconf");
+    args.add("mapreduce.task.files.preserve.failedtasks=true");
+    args.add("-jobconf");
+    args.add("stream.tmpdir="+System.getProperty("test.build.data","/tmp"));
+    args.add("-inputformat");args.add(TextInputFormat.class.getName());
+    super.testCommandLine();
+  }
 
   @Override
   @Test

hadoop-tools/hadoop-streaming/src/test/java/org/apache/hadoop/streaming/TrApp.java

@@ -43,7 +43,6 @@ public class TrApp
     // Note the dots translated to underscore: 
     // property names have been escaped in PipeMapRed.safeEnvVarName()
     expectDefined("mapreduce_cluster_local_dir");
-    expect("mapred_output_format_class", "org.apache.hadoop.mapred.TextOutputFormat");
     expect("mapreduce_map_output_key_class", "org.apache.hadoop.io.Text");
     expect("mapreduce_map_output_value_class", "org.apache.hadoop.io.Text");