HADOOP-1952. Improve handling of invalid, user-specified classes while configuring streaming jobs such as combiner, input/output formats etc. Now invalid options are caught, logged and jobs are failed early. Contributed by Lohit Vijayarenu.

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

CHANGES.txt

@@ -79,6 +79,11 @@ Trunk (unreleased changes)
     HADOOP-1348.  Allow XML comments inside configuration files. 
     (Rajagopal Natarajan and Enis Soztutar via enis)
 
+    HADOOP-1952.  Improve handling of invalid, user-specified classes while
+    configuring streaming jobs such as combiner, input/output formats etc.
+    Now invalid options are caught, logged and jobs are failed early. (Lohit
+    Vijayarenu via acmurthy)
+
 Release 0.15.1 -
 
   BUG FIXES

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

@@ -693,28 +693,33 @@ public class StreamJob {
       fmt = TextInputFormat.class;
     } else if (inputFormatSpec_ != null) {
       if (inputFormatSpec_.equals(TextInputFormat.class.getName())
-          || inputFormatSpec_.equals(TextInputFormat.class.getCanonicalName())) {
+          || inputFormatSpec_.equals(TextInputFormat.class.getCanonicalName())
+          || inputFormatSpec_.equals(TextInputFormat.class.getSimpleName())) {
         fmt = TextInputFormat.class;
       } else if (inputFormatSpec_.equals(KeyValueTextInputFormat.class
           .getName())
           || inputFormatSpec_.equals(KeyValueTextInputFormat.class
-              .getCanonicalName())) {
+              .getCanonicalName())
+          || inputFormatSpec_.equals(KeyValueTextInputFormat.class.getSimpleName())) {
       } else if (inputFormatSpec_.equals(SequenceFileInputFormat.class
           .getName())
           || inputFormatSpec_
               .equals(org.apache.hadoop.mapred.SequenceFileInputFormat.class
-                  .getCanonicalName())) {
+                  .getCanonicalName())
+          || inputFormatSpec_
+              .equals(org.apache.hadoop.mapred.SequenceFileInputFormat.class.getSimpleName())) {
       } else if (inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
           .getName())
           || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class
-              .getCanonicalName())) {
+              .getCanonicalName())
+          || inputFormatSpec_.equals(SequenceFileAsTextInputFormat.class.getSimpleName())) {
         fmt = SequenceFileAsTextInputFormat.class;
       } else {
         c = StreamUtil.goodClassOrNull(inputFormatSpec_, defaultPackage);
         if (c != null) {
           fmt = c;
         } else {
-
+          fail("-inputformat : class not found : " + inputFormatSpec_);
         }
       }
     } 
@@ -744,7 +749,9 @@ public class StreamJob {
       c = StreamUtil.goodClassOrNull(comCmd_, defaultPackage);
       if (c != null) {
         jobConf_.setCombinerClass(c);
-      } 
+      } else {
+        fail("-combiner : class not found : " + comCmd_);
+      }
     }
 
     boolean reducerNone_ = false;
@@ -791,7 +798,9 @@ public class StreamJob {
       c = StreamUtil.goodClassOrNull(outputFormatSpec_, defaultPackage);
       if (c != null) {
         fmt = c;
-      } 
+      } else {
+        fail("-outputformat : class not found : " + outputFormatSpec_);
+      }
     }
     if (fmt == null) {
       fmt = TextOutputFormat.class;
@@ -802,7 +811,9 @@ public class StreamJob {
       c = StreamUtil.goodClassOrNull(partitionerSpec_, defaultPackage);
       if (c != null) {
         jobConf_.setPartitionerClass(c);
-      } 
+      } else {
+        fail("-partitioner : class not found : " + partitionerSpec_);
+      }
     }
     
     if (numReduceTasksSpec_!= null) {

src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestGzipInput.java

@@ -47,7 +47,6 @@ public class TestGzipInput extends TestStreaming
       "-input", INPUT_FILE.getAbsolutePath(),
       "-output", OUTPUT_DIR.getAbsolutePath(),
       "-mapper", map,
-      "-combiner", combine,
       "-reducer", reduce,
       "-jobconf", "stream.recordreader.compression=gzip"
     };

src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreaming.java

@@ -40,7 +40,6 @@ public class TestStreaming extends TestCase
   protected String map = StreamUtil.makeJavaCommand(TrApp.class, new String[]{".", "\\n"});
   // reduce behave like /usr/bin/uniq. But also prepend lines with R.
   // command-line combiner does not have any effect any more.
-  protected String combine  = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"C"});
   protected String reduce = StreamUtil.makeJavaCommand(UniqApp.class, new String[]{"R"});
   protected String outputExpect = "Rare\t\nRblue\t\nRbunnies\t\nRpink\t\nRred\t\nRroses\t\nRviolets\t\n";
 
@@ -66,7 +65,6 @@ public class TestStreaming extends TestCase
       "-input", INPUT_FILE.getAbsolutePath(),
       "-output", OUTPUT_DIR.getAbsolutePath(),
       "-mapper", map,
-      "-combiner", combine,
       "-reducer", reduce,
       //"-verbose",
       //"-jobconf", "stream.debug=set"

src/contrib/streaming/src/test/org/apache/hadoop/streaming/TestStreamingFailure.java

@@ -47,7 +47,6 @@ public class TestStreamingFailure extends TestStreaming
       "-input", INVALID_INPUT_FILE.getAbsolutePath(),
       "-output", OUTPUT_DIR.getAbsolutePath(),
       "-mapper", map,
-      "-combiner", combine,
       "-reducer", reduce,
       //"-verbose",
       //"-jobconf", "stream.debug=set"