HADOOP-15229. Add FileSystem builder-based openFile() API to match createFile();

S3A to implement S3 Select through this API.

The new openFile() API is asynchronous, and implemented across FileSystem and FileContext.

The MapReduce V2 inputs are moved to this API, and you can actually set must/may
options to pass in.

This is more useful for setting things like s3a seek policy than for S3 select,
as the existing input format/record readers can't handle S3 select output where
the stream is shorter than the file length, and splitting plain text is suboptimal.
Future work is needed there.

In the meantime, any/all filesystem connectors are now free to add their own filesystem-specific
configuration parameters which can be set in jobs and used to set filesystem input stream
options (seek policy, retry, encryption secrets, etc).

Contributed by Steve Loughran

Author: steveloughran

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

@@ -25,12 +25,15 @@
 import java.net.URISyntaxException;
 import java.util.ArrayList;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.List;
 import java.util.Map;
 import java.util.NoSuchElementException;
+import java.util.Set;
 import java.util.StringTokenizer;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 
 import org.apache.hadoop.HadoopIllegalArgumentException;
@@ -41,21 +44,21 @@
 import org.apache.hadoop.fs.Options.ChecksumOpt;
 import org.apache.hadoop.fs.Options.CreateOpts;
 import org.apache.hadoop.fs.Options.Rename;
+import org.apache.hadoop.fs.impl.AbstractFSBuilderImpl;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclStatus;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.token.Token;
+import org.apache.hadoop.util.LambdaUtils;
 import org.apache.hadoop.util.Progressable;
 
 import com.google.common.annotations.VisibleForTesting;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import static org.apache.hadoop.fs.impl.PathCapabilitiesSupport.validatePathCapabilityArgs;
-
 /**
  * This class provides an interface for implementors of a Hadoop file system
  * (analogous to the VFS of Unix). Applications do not access this class;
@@ -68,7 +71,7 @@
  */
 @InterfaceAudience.Public
 @InterfaceStability.Stable
-public abstract class AbstractFileSystem implements PathCapabilities {
+public abstract class AbstractFileSystem {
   static final Logger LOG = LoggerFactory.getLogger(AbstractFileSystem.class);
 
   /** Recording statistics per a file system class. */
@@ -398,8 +401,11 @@ public void checkPath(Path path) {
       thatPort = this.getUriDefaultPort();
     }
     if (thisPort != thatPort) {
-      throw new InvalidPathException("Wrong FS: " + path + ", expected: "
-          + this.getUri());
+      throw new InvalidPathException("Wrong FS: " + path
+          + " and port=" + thatPort
+          + ", expected: "
+          + this.getUri()
+          + " with port=" + thisPort);
     }
   }
 
@@ -848,20 +854,6 @@ public abstract FileStatus getFileStatus(final Path f)
       throws AccessControlException, FileNotFoundException,
       UnresolvedLinkException, IOException;
 
-  /**
-   * Synchronize client metadata state.
-   * <p>
-   * In some FileSystem implementations such as HDFS metadata
-   * synchronization is essential to guarantee consistency of read requests
-   * particularly in HA setting.
-   * @throws IOException
-   * @throws UnsupportedOperationException
-   */
-  public void msync() throws IOException, UnsupportedOperationException {
-    throw new UnsupportedOperationException(getClass().getCanonicalName() +
-        " does not support method msync");
-  }
-
   /**
    * The specification of this method matches that of
    * {@link FileContext#access(Path, FsAction)}
@@ -1343,16 +1335,31 @@ public boolean equals(Object other) {
     return myUri.equals(((AbstractFileSystem) other).myUri);
   }
 
-  public boolean hasPathCapability(final Path path,
-      final String capability)
-      throws IOException {
-    switch (validatePathCapabilityArgs(makeQualified(path), capability)) {
-    case CommonPathCapabilities.FS_SYMLINKS:
-      // delegate to the existing supportsSymlinks() call.
-      return supportsSymlinks();
-    default:
-      // the feature is not implemented.
-      return false;
-    }
+  /**
+   * Open a file with the given set of options.
+   * The base implementation performs a blocking
+   * call to {@link #open(Path, int)}in this call;
+   * the actual outcome is in the returned {@code CompletableFuture}.
+   * This avoids having to create some thread pool, while still
+   * setting up the expectation that the {@code get()} call
+   * is needed to evaluate the result.
+   * @param path path to the file
+   * @param mandatoryKeys set of options declared as mandatory.
+   * @param options options set during the build sequence.
+   * @param bufferSize buffer size
+   * @return a future which will evaluate to the opened file.
+   * @throws IOException failure to resolve the link.
+   * @throws IllegalArgumentException unknown mandatory key
+   */
+  public CompletableFuture<FSDataInputStream> openFileWithOptions(Path path,
+      Set<String> mandatoryKeys,
+      Configuration options,
+      int bufferSize) throws IOException {
+    AbstractFSBuilderImpl.rejectUnknownMandatoryKeys(mandatoryKeys,
+        Collections.emptySet(),
+        "for " + path);
+    return LambdaUtils.eval(
+        new CompletableFuture<>(), () -> open(path, bufferSize));
   }
+
 }

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

@@ -24,6 +24,8 @@
 import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.List;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
@@ -262,10 +264,21 @@ public List<Token<?>> getDelegationTokens(String renewer) throws IOException {
     return Arrays.asList(fsImpl.addDelegationTokens(renewer, null));
   }
 
-  @Override
-  public boolean hasPathCapability(final Path path,
-      final String capability)
-      throws IOException {
-    return fsImpl.hasPathCapability(path, capability);
+  /**
+   * Open a file by delegating to
+   * {@link FileSystem#openFileWithOptions(Path, Set, Configuration, int)}.
+   * @param path path to the file
+   * @param mandatoryKeys set of options declared as mandatory.
+   * @param options options set during the build sequence.
+   * @param bufferSize buffer size
+   * @return a future which will evaluate to the opened file.
+   * @throws IOException failure to resolve the link.
+   * @throws IllegalArgumentException unknown mandatory key
+   */
+  public CompletableFuture<FSDataInputStream> openFileWithOptions(Path path,
+      Set<String> mandatoryKeys,
+      Configuration options,
+      int bufferSize) throws IOException {
+    return fsImpl.openFileWithOptions(path, mandatoryKeys, options, bufferSize);
   }
 }

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

@@ -0,0 +1,131 @@
+/*
+ * 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.fs;
+
+import javax.annotation.Nonnull;
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * The base interface which various FileSystem FileContext Builder
+ * interfaces can extend, and which underlying implementations
+ * will then implement.
+ * @param <S> Return type on the {@link #build()} call.
+ * @param <B> type of builder itself.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Unstable
+public interface FSBuilder<S, B extends FSBuilder<S, B>> {
+
+  /**
+   * Set optional Builder parameter.
+   */
+  B opt(@Nonnull String key, @Nonnull String value);
+
+  /**
+   * Set optional boolean parameter for the Builder.
+   *
+   * @see #opt(String, String)
+   */
+  B opt(@Nonnull String key, boolean value);
+
+  /**
+   * Set optional int parameter for the Builder.
+   *
+   * @see #opt(String, String)
+   */
+  B opt(@Nonnull String key, int value);
+
+  /**
+   * Set optional float parameter for the Builder.
+   *
+   * @see #opt(String, String)
+   */
+  B opt(@Nonnull String key, float value);
+
+  /**
+   * Set optional double parameter for the Builder.
+   *
+   * @see #opt(String, String)
+   */
+  B opt(@Nonnull String key, double value);
+
+  /**
+   * Set an array of string values as optional parameter for the Builder.
+   *
+   * @see #opt(String, String)
+   */
+  B opt(@Nonnull String key, @Nonnull String... values);
+
+  /**
+   * Set mandatory option to the Builder.
+   *
+   * If the option is not supported or unavailable,
+   * the client should expect {@link #build()} throws IllegalArgumentException.
+   */
+  B must(@Nonnull String key, @Nonnull String value);
+
+  /**
+   * Set mandatory boolean option.
+   *
+   * @see #must(String, String)
+   */
+  B must(@Nonnull String key, boolean value);
+
+  /**
+   * Set mandatory int option.
+   *
+   * @see #must(String, String)
+   */
+  B must(@Nonnull String key, int value);
+
+  /**
+   * Set mandatory float option.
+   *
+   * @see #must(String, String)
+   */
+  B must(@Nonnull String key, float value);
+
+  /**
+   * Set mandatory double option.
+   *
+   * @see #must(String, String)
+   */
+  B must(@Nonnull String key, double value);
+
+  /**
+   * Set a string array as mandatory option.
+   *
+   * @see #must(String, String)
+   */
+  B must(@Nonnull String key, @Nonnull String... values);
+
+  /**
+   * Instantiate the object which was being built.
+   *
+   * @throws IllegalArgumentException if the parameters are not valid.
+   * @throws UnsupportedOperationException if the filesystem does not support
+   * the specific operation.
+   * @throws IOException on filesystem IO errors.
+   */
+  S build() throws IllegalArgumentException,
+      UnsupportedOperationException, IOException;
+}