HADOOP-1663. Fix streaming to return a non-zero exit code when it fails.

Contributed by Lohit Renu.


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

CHANGES.txt

@@ -529,6 +529,8 @@ Branch 0.14 (unreleased changes)
 155. HADOOP-1714.  Fix TestDFSUpgradeFromImage to work on Windows.
      (Raghu Angadi via nigel)
 
+156. HADOOP-1663.  Return a non-zero exit code if streaming fails. (Lohit Renu
+     via omalley)
 
 Release 0.13.0 - 2007-06-08
 

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

@@ -28,7 +28,12 @@ public class HadoopStreaming {
 
   public static void main(String[] args) throws IOException {
     boolean mayExit = true;
+    int returnStatus = 0;
     StreamJob job = new StreamJob(args, mayExit);
-    job.go();
+    returnStatus = job.go();
+    if (returnStatus != 0) {
+      System.err.println("Streaming Job Failed!");
+      System.exit(returnStatus);
+    }
   }
 }

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

@@ -98,7 +98,7 @@ public class StreamJob {
    * to the jobtracker
    * @throws IOException
    */
-  public void go() throws IOException {
+  public int go() throws IOException {
     init();
 
     preProcessArgs();
@@ -106,7 +106,7 @@ public class StreamJob {
     postProcessArgs();
 
     setJobConf();
-    submitAndMonitorJob();
+    return submitAndMonitorJob();
   }
   
   protected void init() {
@@ -868,7 +868,7 @@ public class StreamJob {
   }
 
   // Based on JobClient
-  public void submitAndMonitorJob() throws IOException {
+  public int submitAndMonitorJob() throws IOException {
 
     if (jar_ != null && isLocalHadoop()) {
       // getAbs became required when shell and subvm have different working dirs...
@@ -906,28 +906,33 @@ public class StreamJob {
       }
       if (!running_.isSuccessful()) {
         jobInfo();
-        throw new IOException("Job not Successful!");
+	LOG.error("Job not Successful!");
+	return 1;
       }
       LOG.info("Job complete: " + jobId_);
       LOG.info("Output: " + output_);
       error = false;
-    } catch(FileNotFoundException fe){
+    } catch(FileNotFoundException fe) {
       LOG.error("Error launching job , bad input path : " + fe.getMessage());
-    }catch(InvalidJobConfException je){
+      return 2;
+    } catch(InvalidJobConfException je) {
       LOG.error("Error launching job , Invalid job conf : " + je.getMessage());
-    }catch(FileAlreadyExistsException fae){
+      return 3;
+    } catch(FileAlreadyExistsException fae) {
       LOG.error("Error launching job , Output path already exists : " 
                 + fae.getMessage());
-    }catch(IOException ioe){
+      return 4;
+    } catch(IOException ioe) {
       LOG.error("Error Launching job : " + ioe.getMessage());
-    }
-    finally {
+      return 5;
+    } finally {
       if (error && (running_ != null)) {
         LOG.info("killJob...");
         running_.killJob();
       }
       jc_.close();
     }
+    return 0;
   }
   /** Support -jobconf x=y x1=y1 type options **/
   class MultiPropertyOption extends PropertyOption{

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

@@ -0,0 +1,86 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.streaming;
+
+import junit.framework.TestCase;
+import java.io.*;
+import java.util.*;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+/**
+ * This class tests if hadoopStreaming returns Exception 
+ * on failure when submitted an invalid/failed job
+ * The test case provides an invalid input file for map/reduce job as
+ * a unit test case
+ */
+public class TestStreamingFailure extends TestStreaming
+{
+
+  protected File INVALID_INPUT_FILE;// = new File("invalid_input.txt");
+  private StreamJob job;
+
+  public TestStreamingFailure() throws IOException
+  {
+    INVALID_INPUT_FILE = new File("invalid_input.txt");
+  }
+
+  protected String[] genArgs() {
+    return new String[] {
+      "-input", INVALID_INPUT_FILE.getAbsolutePath(),
+      "-output", OUTPUT_DIR.getAbsolutePath(),
+      "-mapper", map,
+      "-combiner", combine,
+      "-reducer", reduce,
+      //"-verbose",
+      //"-jobconf", "stream.debug=set"
+      "-jobconf", "keep.failed.task.files=true",
+      "-jobconf", "stream.tmpdir="+System.getProperty("test.build.data","/tmp")
+    };
+  }
+
+  public void testCommandLine()
+  {
+    try {
+      try {
+        OUTPUT_DIR.getAbsoluteFile().delete();
+      } catch (Exception e) {
+      }
+
+      boolean mayExit = false;
+      int returnStatus = 0;
+
+      // During tests, the default Configuration will use a local mapred
+      // So don't specify -config or -cluster
+      job = new StreamJob(genArgs(), mayExit);      
+      returnStatus = job.go();
+      assertEquals("Streaming Job Failure code expected", 5, returnStatus);
+    } catch(Exception e) {
+      // Expecting an exception
+    } finally {
+      OUTPUT_DIR.getAbsoluteFile().delete();
+    }
+  }
+
+  public static void main(String[]args) throws Exception
+  {
+      new TestStreamingFailure().testCommandLine();
+  }
+}