Move checking of output directory existence from JobClient to OutputFormat, so that it can be overridden. Add a base class for OutputFormat that implements this new method.

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

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

@@ -256,15 +256,8 @@ public class JobClient implements MRConstants {
           job.setWorkingDirectory(fileSys.getWorkingDirectory().toString());          
         }
 
-        // Ensure that the output directory is set and not already there
-        File outDir = job.getOutputDir();
-        if (outDir == null && job.getNumReduceTasks() != 0) {
-            throw new IOException("Output directory not set in JobConf.");
-        }
-        if (outDir != null && fs.exists(outDir)) {
-            throw new IOException("Output directory " + outDir + 
-                                  " already exists.");
-        }
+        // Check the output specification
+        job.getOutputFormat().checkOutputSpecs(fs, job);
 
         // Write job file to JobTracker's fs        
         FSDataOutputStream out = fileSys.create(submitJobFile);

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

@@ -28,7 +28,7 @@ import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.conf.Configuration;
 
 /** An {@link OutputFormat} that writes {@link MapFile}s. */
-public class MapFileOutputFormat implements OutputFormat {
+public class MapFileOutputFormat extends OutputFormatBase {
 
   public RecordWriter getRecordWriter(FileSystem fs, JobConf job,
                                       String name) throws IOException {

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

@@ -33,5 +33,15 @@ public interface OutputFormat {
    */
   RecordWriter getRecordWriter(FileSystem fs, JobConf job, String name)
     throws IOException;
+
+  /** Check whether the output specification for a job is appropriate.  Called
+   * when a job is submitted.  Typically checks that it does not already exist,
+   * throwing an exception when it already exists, so that output is not
+   * overwritten.
+   *
+   * @param job the job whose output will be written
+   * @throws IOException when output should not be attempted
+   */
+  void checkOutputSpecs(FileSystem fs, JobConf job) throws IOException;
 }
 

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

@@ -0,0 +1,43 @@
+/**
+ * Copyright 2006 The Apache Software Foundation
+ *
+ * Licensed 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.mapred;
+
+import java.io.IOException;
+import java.io.File;
+
+import org.apache.hadoop.fs.FileSystem;
+
+/** A base class for {@link OutputFormat}. */
+public abstract class OutputFormatBase implements OutputFormat {
+  public abstract RecordWriter getRecordWriter(FileSystem fs,
+                                               JobConf job, String name)
+    throws IOException;
+
+  public void checkOutputSpecs(FileSystem fs, JobConf job) throws IOException {
+    // Ensure that the output directory is set and not already there
+    File outDir = job.getOutputDir();
+    if (outDir == null && job.getNumReduceTasks() != 0) {
+      throw new IOException("Output directory not set in JobConf.");
+    }
+    if (outDir != null && fs.exists(outDir)) {
+      throw new IOException("Output directory " + outDir + 
+                            " already exists.");
+    }
+  }
+
+}
+

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

@@ -28,7 +28,7 @@ import org.apache.hadoop.io.Writable;
 import org.apache.hadoop.conf.Configuration;
 
 /** An {@link OutputFormat} that writes {@link SequenceFile}s. */
-public class SequenceFileOutputFormat implements OutputFormat {
+public class SequenceFileOutputFormat extends OutputFormatBase {
 
   public RecordWriter getRecordWriter(FileSystem fs, JobConf job,
                                       String name) throws IOException {

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

@@ -26,7 +26,7 @@ import org.apache.hadoop.io.WritableComparable;
 import org.apache.hadoop.io.Writable;
 
 /** An {@link OutputFormat} that writes plain text files. */
-public class TextOutputFormat implements OutputFormat {
+public class TextOutputFormat extends OutputFormatBase {
 
   public RecordWriter getRecordWriter(FileSystem fs, JobConf job,
                                       String name) throws IOException {