HADOOP-19354. S3A: Create S3 InputStreams via a factory

S3 InputStreams are created by a factory class, with the
choice of factory dynamically chosen by the option

  fs.s3a.input.stream.type

Supported values: classic, prefetching, analytics.

S3AStore

* Manages the creation and service lifecycle of the chosen factory,
  as well as forwarding stream construction requests to the chosen factory.
* Provides the callbacks needed by both the factories and input streams.
* StreamCapabilities.hasCapability(), which is
  relayed to the active factory. This avoids the FS having
  to know what capabilities are available in the stream.

Author: steveloughran

hadoop-tools/hadoop-aws/dev-support/findbugs-exclude.xml

@@ -30,7 +30,7 @@
   </Match>
   <!-- we are using completable futures, so ignore the Future which submit() returns -->
   <Match>
-    <Class name="org.apache.hadoop.fs.s3a.S3AFileSystem$InputStreamCallbacksImpl" />
+    <Class name="org.apache.hadoop.fs.s3a.impl.InputStreamCallbacksImpl" />
     <Bug pattern="RV_RETURN_VALUE_IGNORED_BAD_PRACTICE" />
   </Match>
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/Constants.java

@@ -21,6 +21,8 @@
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.Options;
+import org.apache.hadoop.fs.s3a.impl.streams.InputStreamType;
+import org.apache.hadoop.fs.s3a.impl.streams.StreamIntegration;
 import org.apache.hadoop.security.ssl.DelegatingSSLSocketFactory;
 
 import java.time.Duration;
@@ -1580,6 +1582,34 @@ private Constants() {
    */
   public static final String AWS_AUTH_CLASS_PREFIX = "com.amazonaws.auth";
 
+  /**
+   * Input stream type: {@value}.
+   */
+  public static final String INPUT_STREAM_TYPE = "fs.s3a.input.stream.type";
+
+  /**
+   * The classic input stream: {@value}.
+   */
+  public static final String INPUT_STREAM_TYPE_CLASSIC =
+      StreamIntegration.CLASSIC;
+
+  /**
+   * The prefetching input stream: "prefetch".
+   */
+  public static final String INPUT_STREAM_TYPE_PREFETCH = StreamIntegration.PREFETCH;
+
+  /**
+   * The analytics input stream: "analytics".
+   */
+  public static final String INPUT_STREAM_TYPE_ANALYTICS =
+      StreamIntegration.ANALYTICS;
+
+  /**
+   * The default input stream.
+   * Currently {@link #INPUT_STREAM_TYPE_CLASSIC}
+   */
+  public static final String INPUT_STREAM_TYPE_DEFAULT = InputStreamType.DEFAULT_STREAM_TYPE.getName();
+
   /**
    * Controls whether the prefetching input stream is enabled.
    */