HADOOP-14396. Add builder interface to FileContext.

Contributed by  Lei (Eddy) Xu.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FSDataOutputStreamBuilder.java

@@ -115,6 +115,27 @@ public abstract class FSDataOutputStreamBuilder
    */
   protected abstract B getThisBuilder();
 
+  /**
+   * Construct from a {@link FileContext}.
+   *
+   * @param fc FileContext
+   * @param p path.
+   * @throws IOException
+   */
+  FSDataOutputStreamBuilder(@Nonnull FileContext fc,
+      @Nonnull Path p) throws IOException {
+    Preconditions.checkNotNull(fc);
+    Preconditions.checkNotNull(p);
+    this.fs = null;
+    this.path = p;
+
+    AbstractFileSystem afs = fc.getFSofPath(p);
+    FsServerDefaults defaults = afs.getServerDefaults(p);
+    bufferSize = defaults.getFileBufferSize();
+    replication = defaults.getReplication();
+    blockSize = defaults.getBlockSize();
+  }
+
   /**
    * Constructor.
    */
@@ -131,6 +152,7 @@ public abstract class FSDataOutputStreamBuilder
   }
 
   protected FileSystem getFS() {
+    Preconditions.checkNotNull(fs);
     return fs;
   }
 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/FileContext.java

@@ -24,6 +24,7 @@ import java.io.OutputStream;
 import java.net.URI;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.HashSet;
@@ -35,6 +36,8 @@ import java.util.Stack;
 import java.util.TreeSet;
 import java.util.Map.Entry;
 
+import javax.annotation.Nonnull;
+
 import org.apache.hadoop.HadoopIllegalArgumentException;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -694,6 +697,69 @@ public class FileContext {
     }.resolve(this, absF);
   }
 
+  /**
+   * {@link FSDataOutputStreamBuilder} for {@liink FileContext}.
+   */
+  private static final class FCDataOutputStreamBuilder extends
+      FSDataOutputStreamBuilder<
+        FSDataOutputStream, FCDataOutputStreamBuilder> {
+    private final FileContext fc;
+
+    private FCDataOutputStreamBuilder(
+        @Nonnull FileContext fc, @Nonnull Path p) throws IOException {
+      super(fc, p);
+      this.fc = fc;
+      Preconditions.checkNotNull(fc);
+    }
+
+    @Override
+    protected FCDataOutputStreamBuilder getThisBuilder() {
+      return this;
+    }
+
+    @Override
+    public FSDataOutputStream build() throws IOException {
+      final EnumSet<CreateFlag> flags = getFlags();
+      List<CreateOpts> createOpts = new ArrayList<>(Arrays.asList(
+          CreateOpts.blockSize(getBlockSize()),
+          CreateOpts.bufferSize(getBufferSize()),
+          CreateOpts.repFac(getReplication()),
+          CreateOpts.perms(getPermission())
+      ));
+      if (getChecksumOpt() != null) {
+        createOpts.add(CreateOpts.checksumParam(getChecksumOpt()));
+      }
+      if (getProgress() != null) {
+        createOpts.add(CreateOpts.progress(getProgress()));
+      }
+      if (isRecursive()) {
+        createOpts.add(CreateOpts.createParent());
+      }
+      return fc.create(getPath(), flags,
+          createOpts.toArray(new CreateOpts[0]));
+    }
+  }
+
+  /**
+   * Create a {@link FSDataOutputStreamBuilder} for creating or overwriting
+   * a file on indicated path.
+   *
+   * @param f the file path to create builder for.
+   * @return {@link FSDataOutputStreamBuilder} to build a
+   *         {@link FSDataOutputStream}.
+   *
+   * Upon {@link FSDataOutputStreamBuilder#build()} being invoked,
+   * builder parameters will be verified by {@link FileContext} and
+   * {@link AbstractFileSystem#create}. And filesystem states will be modified.
+   *
+   * Client should expect {@link FSDataOutputStreamBuilder#build()} throw the
+   * same exceptions as create(Path, EnumSet, CreateOpts...).
+   */
+  public FSDataOutputStreamBuilder<FSDataOutputStream, ?> create(final Path f)
+      throws IOException {
+    return new FCDataOutputStreamBuilder(this, f).create();
+  }
+
   /**
    * Make(create) a directory and all the non-existent parents.
    * 

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/Options.java

@@ -55,6 +55,9 @@ public final class Options {
         ChecksumOpt csumOpt) {
       return new ChecksumParam(csumOpt);
     }
+    public static Progress progress(Progressable prog) {
+      return new Progress(prog);
+    }
     public static Perms perms(FsPermission perm) {
       return new Perms(perm);
     }

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/fs/FileContextMainOperationsBaseTest.java

@@ -810,7 +810,49 @@ public abstract class FileContextMainOperationsBaseTest  {
     fc.create(p, EnumSet.of(CREATE, APPEND, OVERWRITE));
     Assert.fail("Excepted exception not thrown");
   }
-  
+
+  @Test
+  public void testBuilderCreateNonExistingFile() throws IOException {
+    Path p = getTestRootPath(fc, "test/testBuilderCreateNonExistingFile");
+    FSDataOutputStream out = fc.create(p).build();
+    writeData(fc, p, out, data, data.length);
+  }
+
+  @Test
+  public void testBuilderCreateExistingFile() throws IOException {
+    Path p = getTestRootPath(fc, "test/testBuilderCreateExistingFile");
+    createFile(p);
+    FSDataOutputStream out = fc.create(p).overwrite(true).build();
+    writeData(fc, p, out, data, data.length);
+  }
+
+  @Test
+  public void testBuilderCreateAppendNonExistingFile() throws IOException {
+    Path p = getTestRootPath(fc, "test/testBuilderCreateAppendNonExistingFile");
+    FSDataOutputStream out = fc.create(p).append().build();
+    writeData(fc, p, out, data, data.length);
+  }
+
+  @Test
+  public void testBuilderCreateAppendExistingFile() throws IOException {
+    Path p = getTestRootPath(fc, "test/testBuilderCreateAppendExistingFile");
+    createFile(p);
+    FSDataOutputStream out = fc.create(p).append().build();
+    writeData(fc, p, out, data, 2 * data.length);
+  }
+
+  @Test
+  public void testBuilderCreateRecursive() throws IOException {
+    Path p = getTestRootPath(fc, "test/parent/no/exist/file1");
+    try (FSDataOutputStream out = fc.create(p).build()) {
+      fail("Should throw FileNotFoundException on non-exist directory");
+    } catch (FileNotFoundException e) {
+    }
+
+    FSDataOutputStream out = fc.create(p).recursive().build();
+    writeData(fc, p, out, data, data.length);
+  }
+
   private static void writeData(FileContext fc, Path p, FSDataOutputStream out,
       byte[] data, long expectedLen) throws IOException {
     out.write(data, 0, data.length);