Merge -r 638745:638746 from trunk to 0.16. Fixes HADOOP-2985.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.16@638748 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -14,6 +14,10 @@ Release 0.16.2 - Unreleased
     HADOOP-2978. Fixes the JobHistory log format for counters.
     (Runping Qi via ddas)
 
+    HADOOP-2985. Fixes LocalJobRunner to tolerate null job output path.
+    Also makes the _temporary a constant in MRConstants.java.
+    (Amareshwari Sriramadasu via ddas)
+
 Release 0.16.1 - 2008-03-13
 
   INCOMPATIBLE CHANGES

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

@@ -279,13 +279,11 @@ class JobInProgress {
     // create job specific temporary directory in output path
     Path outputPath = conf.getOutputPath();
     if (outputPath != null) {
-      Path tmpDir = new Path(outputPath, "_temporary");
+      Path tmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
       FileSystem fileSys = tmpDir.getFileSystem(conf);
       if (!fileSys.mkdirs(tmpDir)) {
         LOG.error("Mkdirs failed to create " + tmpDir.toString());
       }
-    } else {
-      LOG.error("Null Output path");
     }
 
     this.status = new JobStatus(status.getJobId(), 0.0f, 0.0f, JobStatus.RUNNING);
@@ -1145,7 +1143,7 @@ class JobInProgress {
       // delete the temporary directory in output directory
       Path outputPath = conf.getOutputPath();
       if (outputPath != null) {
-        Path tmpDir = new Path(outputPath, "_temporary");
+        Path tmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
         FileSystem fileSys = tmpDir.getFileSystem(conf);
         if (fileSys.exists(tmpDir)) {
           FileUtil.fullyDelete(fileSys, tmpDir);

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

@@ -114,10 +114,15 @@ class LocalJobRunner implements JobSubmissionProtocol {
           job.setNumReduceTasks(1);
         }
         // create job specific temp directory in output path
-        Path tmpDir = new Path(job.getOutputPath(), "_temporary");
-        FileSystem fileSys = tmpDir.getFileSystem(job);
-        if (!fileSys.mkdirs(tmpDir)) {
-          LOG.error("Mkdirs failed to create " + tmpDir.toString());
+        Path outputPath = job.getOutputPath();
+        FileSystem outputFs = null;
+        Path tmpDir = null;
+        if (outputPath != null) {
+          tmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
+          outputFs = tmpDir.getFileSystem(job);
+          if (!outputFs.mkdirs(tmpDir)) {
+            LOG.error("Mkdirs failed to create " + tmpDir.toString());
+          }
         }
 
         DataOutputBuffer buffer = new DataOutputBuffer();
@@ -133,15 +138,17 @@ class LocalJobRunner implements JobSubmissionProtocol {
                                     splits[i].getClass().getName(),
                                     split);
           JobConf localConf = new JobConf(job);
-          if (fileSys.exists(tmpDir)) {
-            Path taskTmpDir = new Path(tmpDir, "_" + mapId);
-            if (!fileSys.mkdirs(taskTmpDir)) {
-              throw new IOException("Mkdirs failed to create " 
-                                     + taskTmpDir.toString());
-            }
-          } else {
-            throw new IOException("The directory " + tmpDir.toString()
+          if (outputFs != null) {
+            if (outputFs.exists(tmpDir)) {
+              Path taskTmpDir = new Path(tmpDir, "_" + mapId);
+              if (!outputFs.mkdirs(taskTmpDir)) {
+                throw new IOException("Mkdirs failed to create " 
+                                       + taskTmpDir.toString());
+              }
+            } else {
+              throw new IOException("The directory " + tmpDir.toString()
                                    + " doesnt exist " );
+            }
           }
           map.localizeConfiguration(localConf);
           map.setConf(localConf);
@@ -175,15 +182,17 @@ class LocalJobRunner implements JobSubmissionProtocol {
                                                  "tip_r_0001",
                                                  reduceId, 0, mapIds.size());
               JobConf localConf = new JobConf(job);
-              if (fileSys.exists(tmpDir)) {
-                Path taskTmpDir = new Path(tmpDir, "_" + reduceId);
-                if (!fileSys.mkdirs(taskTmpDir)) {
-                  throw new IOException("Mkdirs failed to create " 
-                                         + taskTmpDir.toString());
+              if (outputFs != null) {
+                if (outputFs.exists(tmpDir)) {
+                  Path taskTmpDir = new Path(tmpDir, "_" + reduceId);
+                  if (!outputFs.mkdirs(taskTmpDir)) {
+                    throw new IOException("Mkdirs failed to create " 
+                                           + taskTmpDir.toString());
+                  }
+                } else {
+                  throw new IOException("The directory " + tmpDir.toString()
+                                       + " doesnt exist ");
                 }
-              } else {
-                throw new IOException("The directory " + tmpDir.toString()
-                                       + " doesnt exist "); 
               }
               reduce.localizeConfiguration(localConf);
               reduce.setConf(localConf);
@@ -207,8 +216,10 @@ class LocalJobRunner implements JobSubmissionProtocol {
         }
         // delete the temporary directory in output directory
         try {
-          if (fileSys.exists(tmpDir)) {
-            FileUtil.fullyDelete(fileSys, tmpDir);
+          if (outputFs != null) {
+            if (outputFs.exists(tmpDir)) {
+              FileUtil.fullyDelete(outputFs, tmpDir);
+            }
           }
         } catch (IOException e) {
           LOG.error("Exception in deleting " + tmpDir.toString());

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

@@ -54,4 +54,9 @@ interface MRConstants {
    * The custom http header used for the map output length.
    */
   public static final String MAP_OUTPUT_LENGTH = "Map-Output-Length";
+
+  /**
+   * Temporary directory name 
+   */
+  public static final String TEMP_DIR_NAME = "_temporary";
 }

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

@@ -190,8 +190,8 @@ abstract class Task implements Writable, Configurable {
   public String toString() { return taskId; }
 
   private Path getTaskOutputPath(JobConf conf) {
-    Path p = new Path(conf.getOutputPath(), ("_temporary" 
-                      + Path.SEPARATOR + "_" + taskId));
+    Path p = new Path(conf.getOutputPath(), 
+      (MRConstants.TEMP_DIR_NAME + Path.SEPARATOR + "_" + taskId));
     try {
       FileSystem fs = p.getFileSystem(conf);
       return p.makeQualified(fs);

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

@@ -1422,7 +1422,7 @@ public class TaskTracker
       // create _taskid directory in output path temporary directory.
       Path outputPath = localJobConf.getOutputPath();
       if (outputPath != null) {
-        Path jobTmpDir = new Path(outputPath, "_temporary");
+        Path jobTmpDir = new Path(outputPath, MRConstants.TEMP_DIR_NAME);
         FileSystem fs = jobTmpDir.getFileSystem(localJobConf);
         if (fs.exists(jobTmpDir)) {
           Path taskTmpDir = new Path(jobTmpDir, "_" + task.getTaskId());