HADOOP-2130. Pipes submit job should have both blocking and non-blocking

versions. (acmurthy via omalley)


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

CHANGES.txt

@@ -222,6 +222,9 @@ Trunk (unreleased changes)
 
     HADOOP-3816. Faster directory listing in KFS. (Sriram Rao via omalley)
 
+    HADOOP-2130. Pipes submit job should have both blocking and non-blocking
+    versions. (acmurthy via omalley)
+
   BUG FIXES
 
     HADOOP-3563.  Refactor the distributed upgrade code so that it is 

src/mapred/org/apache/hadoop/mapred/pipes/Submitter.java

@@ -209,8 +209,40 @@ public class Submitter {
    * to the job to run under pipes are made to the configuration.
    * @param conf the job to submit to the cluster (MODIFIED)
    * @throws IOException
+   * @deprecated Use {@link Submitter#runJob(JobConf)}
    */
+  @Deprecated
   public static RunningJob submitJob(JobConf conf) throws IOException {
+    return runJob(conf);
+  }
+
+  /**
+   * Submit a job to the map/reduce cluster. All of the necessary modifications
+   * to the job to run under pipes are made to the configuration.
+   * @param conf the job to submit to the cluster (MODIFIED)
+   * @throws IOException
+   */
+  public static RunningJob runJob(JobConf conf) throws IOException {
+    setupPipesJob(conf);
+    return JobClient.runJob(conf);
+  }
+
+  /**
+   * Submit a job to the Map-Reduce framework.
+   * This returns a handle to the {@link RunningJob} which can be used to track
+   * the running-job.
+   * 
+   * @param conf the job configuration.
+   * @return a handle to the {@link RunningJob} which can be used to track the
+   *         running-job.
+   * @throws IOException
+   */
+  public static RunningJob jobSubmit(JobConf conf) throws IOException {
+    setupPipesJob(conf);
+    return new JobClient(conf).submitJob(conf);
+  }
+  
+  private static void setupPipesJob(JobConf conf) throws IOException {
     // default map output types to Text
     if (!getIsJavaMapper(conf)) {
       conf.setMapRunnerClass(PipesMapRunner.class);
@@ -258,7 +290,6 @@ public class Submitter {
       throw ie;
     }
     DistributedCache.setCacheFiles(fileCache, conf);
-    return JobClient.runJob(conf);
   }
 
   /**
@@ -414,7 +445,7 @@ public class Submitter {
                                         pathToFile(new Path(jarFile)).toURL()});
         conf.setClassLoader(loader);
       }
-      submitJob(conf);
+      runJob(conf);
     } catch (OptionException oe) {
       cli.printUsage();
     }

src/test/org/apache/hadoop/mapred/pipes/TestPipes.java

@@ -149,8 +149,21 @@ public class TestPipes extends TestCase {
       Submitter.setIsJavaRecordWriter(job, true);
       FileInputFormat.setInputPaths(job, inputPath);
       FileOutputFormat.setOutputPath(job, outputPath);
-      RunningJob result = Submitter.submitJob(job);
-      assertTrue("pipes job failed", result.isSuccessful());
+      RunningJob rJob = null;
+      if (numReduces == 0) {
+        rJob = Submitter.jobSubmit(job);
+        
+        while (!rJob.isComplete()) {
+          try {
+            Thread.sleep(1000);
+          } catch (InterruptedException ie) {
+            throw new RuntimeException(ie);
+          }
+        }
+      } else {
+        rJob = Submitter.runJob(job);
+      }
+      assertTrue("pipes job failed", rJob.isSuccessful());
     }
 
     List<String> results = new ArrayList<String>();