S3A S3Seekable stream refactor + move S3AInputStream creation to factory under S3AStore

hadoop-hdfs-project/hadoop-hdfs-rbf/src/main/java/org/apache/hadoop/hdfs/server/federation/resolver/order/AvailableSpaceResolver.java

@@ -29,6 +29,7 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hdfs.server.federation.resolver.PathLocation;
+import org.apache.hadoop.hdfs.server.federation.resolver.RemoteLocation;
 import org.apache.hadoop.hdfs.server.federation.resolver.order.AvailableSpaceResolver.SubclusterAvailableSpace;
 import org.apache.hadoop.hdfs.server.federation.router.RBFConfigKeys;
 import org.apache.hadoop.hdfs.server.federation.router.Router;
@@ -116,8 +117,10 @@ protected Map<String, SubclusterAvailableSpace> getSubclusterInfo(
   protected String chooseFirstNamespace(String path, PathLocation loc) {
     Map<String, SubclusterAvailableSpace> subclusterInfo =
         getSubclusterMapping();
-    List<SubclusterAvailableSpace> subclusterList = new LinkedList<>(
-        subclusterInfo.values());
+    List<SubclusterAvailableSpace> subclusterList = new LinkedList<>();
+    for (RemoteLocation dest : loc.getDestinations()) {
+      subclusterList.add(subclusterInfo.get(dest.getNameserviceId()));
+    }
     Collections.sort(subclusterList, comparator);
 
     return subclusterList.size() > 0 ? subclusterList.get(0).getNameserviceId()

hadoop-hdfs-project/hadoop-hdfs-rbf/src/test/java/org/apache/hadoop/hdfs/server/federation/resolver/order/TestAvailableSpaceResolver.java

@@ -229,4 +229,12 @@ private void verifyRank(float balancerPreference, boolean shouldOrdered,
     }
     subclusters.clear();
   }
+
+  @Test
+  public void testChooseFirstNamespace() throws Exception {
+    MultipleDestinationMountTableResolver mountTableResolver =
+            mockAvailableSpaceResolver(1.0f);
+    PathLocation loc = mountTableResolver.getDestinationForPath("/space");
+    assertEquals("subcluster9", loc.getDefaultLocation().getNameserviceId());
+  }
 }

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/pom.xml

@@ -48,10 +48,6 @@
     <!-- Set a longer timeout for integration test (in milliseconds) -->
     <test.integration.timeout>200000</test.integration.timeout>
 
-    <!--   should directory marker retention be audited? -->
-    <fs.s3a.directory.marker.audit>false</fs.s3a.directory.marker.audit>
-    <!--    marker retention policy -->
-    <fs.s3a.directory.marker.retention></fs.s3a.directory.marker.retention>
 
     <!-- Is prefetch enabled? -->
     <fs.s3a.prefetch.enabled>unset</fs.s3a.prefetch.enabled>
@@ -126,9 +122,6 @@
                 <fs.s3a.scale.test.huge.filesize>${fs.s3a.scale.test.huge.filesize}</fs.s3a.scale.test.huge.filesize>
                 <fs.s3a.scale.test.huge.huge.partitionsize>${fs.s3a.scale.test.huge.partitionsize}</fs.s3a.scale.test.huge.huge.partitionsize>
                 <fs.s3a.scale.test.timeout>${fs.s3a.scale.test.timeout}</fs.s3a.scale.test.timeout>
-                <!-- Markers-->
-                <fs.s3a.directory.marker.retention>${fs.s3a.directory.marker.retention}</fs.s3a.directory.marker.retention>
-                <fs.s3a.directory.marker.audit>${fs.s3a.directory.marker.audit}</fs.s3a.directory.marker.audit>
                 <!-- Prefetch -->
                 <fs.s3a.prefetch.enabled>${fs.s3a.prefetch.enabled}</fs.s3a.prefetch.enabled>
               </systemPropertyVariables>
@@ -167,8 +160,6 @@
                     <fs.s3a.scale.test.huge.filesize>${fs.s3a.scale.test.huge.filesize}</fs.s3a.scale.test.huge.filesize>
                     <fs.s3a.scale.test.huge.huge.partitionsize>${fs.s3a.scale.test.huge.partitionsize}</fs.s3a.scale.test.huge.huge.partitionsize>
                     <fs.s3a.scale.test.timeout>${fs.s3a.scale.test.timeout}</fs.s3a.scale.test.timeout>
-                    <fs.s3a.directory.marker.retention>${fs.s3a.directory.marker.retention}</fs.s3a.directory.marker.retention>
-
                     <test.default.timeout>${test.integration.timeout}</test.default.timeout>
                     <!-- Prefetch -->
                     <fs.s3a.prefetch.enabled>${fs.s3a.prefetch.enabled}</fs.s3a.prefetch.enabled>
@@ -221,9 +212,6 @@
                     <fs.s3a.scale.test.huge.filesize>${fs.s3a.scale.test.huge.filesize}</fs.s3a.scale.test.huge.filesize>
                     <fs.s3a.scale.test.huge.huge.partitionsize>${fs.s3a.scale.test.huge.partitionsize}</fs.s3a.scale.test.huge.huge.partitionsize>
                     <fs.s3a.scale.test.timeout>${fs.s3a.scale.test.timeout}</fs.s3a.scale.test.timeout>
-                    <!-- Markers-->
-                    <fs.s3a.directory.marker.retention>${fs.s3a.directory.marker.retention}</fs.s3a.directory.marker.retention>
-                    <fs.s3a.directory.marker.audit>${fs.s3a.directory.marker.audit}</fs.s3a.directory.marker.audit>
                     <!-- Prefetch -->
                     <fs.s3a.prefetch.enabled>${fs.s3a.prefetch.enabled}</fs.s3a.prefetch.enabled>
                     <!-- are root tests enabled. Set to false when running parallel jobs on same bucket -->
@@ -285,9 +273,6 @@
                     <fs.s3a.scale.test.enabled>${fs.s3a.scale.test.enabled}</fs.s3a.scale.test.enabled>
                     <fs.s3a.scale.test.huge.filesize>${fs.s3a.scale.test.huge.filesize}</fs.s3a.scale.test.huge.filesize>
                     <fs.s3a.scale.test.timeout>${fs.s3a.scale.test.timeout}</fs.s3a.scale.test.timeout>
-                    <!-- Markers-->
-                    <fs.s3a.directory.marker.retention>${fs.s3a.directory.marker.retention}</fs.s3a.directory.marker.retention>
-                    <fs.s3a.directory.marker.audit>${fs.s3a.directory.marker.audit}</fs.s3a.directory.marker.audit>
                     <!-- Prefetch -->
                     <fs.s3a.prefetch.enabled>${fs.s3a.prefetch.enabled}</fs.s3a.prefetch.enabled>
                     <test.unique.fork.id>job-${job.id}</test.unique.fork.id>
@@ -314,44 +299,6 @@
       </properties>
     </profile>
 
-    <!-- Directory marker retention options, all from the -Dmarkers value-->
-    <profile>
-      <id>keep-markers</id>
-      <activation>
-        <property>
-          <name>markers</name>
-          <value>keep</value>
-        </property>
-      </activation>
-      <properties >
-        <fs.s3a.directory.marker.retention>keep</fs.s3a.directory.marker.retention>
-      </properties>
-    </profile>
-    <profile>
-      <id>delete-markers</id>
-      <activation>
-        <property>
-          <name>markers</name>
-          <value>delete</value>
-        </property>
-      </activation>
-      <properties >
-        <fs.s3a.directory.marker.retention>delete</fs.s3a.directory.marker.retention>
-      </properties>
-    </profile>
-    <profile>
-      <id>auth-markers</id>
-      <activation>
-        <property>
-          <name>markers</name>
-          <value>authoritative</value>
-        </property>
-      </activation>
-      <properties >
-        <fs.s3a.directory.marker.retention>authoritative</fs.s3a.directory.marker.retention>
-      </properties>
-    </profile>
-
     <!-- Turn on prefetching-->
     <profile>
       <id>prefetch</id>
@@ -525,6 +472,24 @@
       <artifactId>amazon-s3-encryption-client-java</artifactId>
       <scope>provided</scope>
     </dependency>
+    <!-- <dependency>
+      <groupId>software.amazon.s3.analyticsaccelerator</groupId>
+      <artifactId>analyticsaccelerator-s3</artifactId>
+      <version>0.0.2</version>
+      <scope>compile</scope>
+    </dependency> -->
+    <dependency>
+      <groupId>software.amazon.s3.analyticsaccelerator</groupId>
+      <artifactId>analyticsaccelerator-s3</artifactId>
+      <version>0.0.2</version>
+      <scope>system</scope>
+      <systemPath>${project.basedir}/libaws/analyticsaccelerator-s3-0.0.2.jar</systemPath>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk.crt</groupId>
+      <artifactId>aws-crt</artifactId>
+      <version>0.29.10</version>
+    </dependency>
     <dependency>
       <groupId>org.assertj</groupId>
       <artifactId>assertj-core</artifactId>

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

@@ -21,6 +21,7 @@
 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.security.ssl.DelegatingSSLSocketFactory;
 
 import java.time.Duration;
@@ -848,6 +849,7 @@ private Constants() {
       "fs.s3a." + Constants.AWS_SERVICE_IDENTIFIER_STS.toLowerCase()
           + ".signing-algorithm";
 
+  @Deprecated
   public static final String S3N_FOLDER_SUFFIX = "_$folder$";
   public static final String FS_S3A_BLOCK_SIZE = "fs.s3a.block.size";
   public static final String FS_S3A = "s3a";
@@ -868,10 +870,13 @@ private Constants() {
   /**
    * Paths considered "authoritative".
    * When S3guard was supported, this skipped checks to s3 on directory listings.
-   * It is also use to optionally disable marker retentation purely on these
-   * paths -a feature which is still retained/available.
+   * It was also possilbe to use to optionally disable marker retentation purely on these
+   * paths -a feature which is no longer available.
+   * As no feature uses this any more, it is declared as deprecated.
    * */
+  @Deprecated
   public static final String AUTHORITATIVE_PATH = "fs.s3a.authoritative.path";
+  @Deprecated
   public static final String[] DEFAULT_AUTHORITATIVE_PATH = {};
 
   /**
@@ -1339,45 +1344,52 @@ private Constants() {
 
   /**
    * Policy for directory markers.
-   * This is a new feature of HADOOP-13230 which addresses
-   * some scale, performance and permissions issues -but
-   * at the risk of backwards compatibility.
+   * No longer supported as "keep" is the sole policy.
    */
+  @Deprecated
   public static final String DIRECTORY_MARKER_POLICY =
       "fs.s3a.directory.marker.retention";
 
   /**
-   * Delete directory markers. This is the backwards compatible option.
+   * Delete directory markers.
+   * No longer supported as "keep" is the sole policy.
    * Value: {@value}.
    */
+  @Deprecated
   public static final String DIRECTORY_MARKER_POLICY_DELETE =
       "delete";
 
   /**
    * Retain directory markers.
+   * No longer needed, so marked as deprecated to flag usages.
    * Value: {@value}.
    */
+  @Deprecated
   public static final String DIRECTORY_MARKER_POLICY_KEEP =
       "keep";
 
   /**
    * Retain directory markers in authoritative directory trees only.
+   * No longer required as "keep" is the sole policy.
    * Value: {@value}.
    */
+  @Deprecated
   public static final String DIRECTORY_MARKER_POLICY_AUTHORITATIVE =
       "authoritative";
 
   /**
    * Default retention policy: {@value}.
+   * No longer required as "keep" is the sole policy.
    */
+  @Deprecated
   public static final String DEFAULT_DIRECTORY_MARKER_POLICY =
       DIRECTORY_MARKER_POLICY_KEEP;
 
 
   /**
    * {@code PathCapabilities} probe to verify that an S3A Filesystem
    * has the changes needed to safely work with buckets where
-   * directoy markers have not been deleted.
+   * directory markers have not been deleted.
    * Value: {@value}.
    */
   public static final String STORE_CAPABILITY_DIRECTORY_MARKER_AWARE
@@ -1394,23 +1406,28 @@ private Constants() {
   /**
    * {@code PathCapabilities} probe to indicate that the filesystem
    * deletes directory markers.
+   * Always false.
    * Value: {@value}.
    */
+  @Deprecated
   public static final String STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_DELETE
       = "fs.s3a.capability.directory.marker.policy.delete";
 
   /**
    * {@code PathCapabilities} probe to indicate that the filesystem
    * keeps directory markers in authoritative paths only.
+   * This probe always returns false.
    * Value: {@value}.
    */
+  @Deprecated
   public static final String
       STORE_CAPABILITY_DIRECTORY_MARKER_POLICY_AUTHORITATIVE =
       "fs.s3a.capability.directory.marker.policy.authoritative";
 
   /**
    * {@code PathCapabilities} probe to indicate that a path
    * keeps directory markers.
+   * This probe always returns true.
    * Value: {@value}.
    */
   public static final String STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_KEEP
@@ -1419,6 +1436,7 @@ private Constants() {
   /**
    * {@code PathCapabilities} probe to indicate that a path
    * deletes directory markers.
+   * This probe always returns false.
    * Value: {@value}.
    */
   public static final String STORE_CAPABILITY_DIRECTORY_MARKER_ACTION_DELETE
@@ -1560,6 +1578,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 =
+      InputStreamType.Classic.getName();
+
+  /**
+   * The prefetching input stream: "prefetch".
+   */
+  public static final String INPUT_STREAM_TYPE_PREFETCH = InputStreamType.Prefetch.getName();
+
+  /**
+   * The analytics input stream: "analytics".
+   */
+  public static final String INPUT_STREAM_TYPE_ANALYTICS =
+      InputStreamType.Analytics.getName();
+
+  /**
+   * 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.
    */
@@ -1760,4 +1806,25 @@ private Constants() {
    * Value: {@value}.
    */
   public static final String S3A_IO_RATE_LIMIT = "fs.s3a.io.rate.limit";
+
+
+  /**
+   * Prefix to configure Analytics Accelerator Library.
+   */
+  public static final String ANALYTICS_ACCELERATOR_CONFIGURATION_PREFIX =
+          "fs.s3a.analytics.accelerator";
+
+  /**
+   * Config to enable usage of crt client with Analytics Accelerator Library.
+   * It is by default true.
+   */
+  public static final String ANALYTICS_ACCELERATOR_CRT_ENABLED =
+          "fs.s3a.analytics.accelerator.crt.client";
+
+  /**
+   * Default value for {@link #ANALYTICS_ACCELERATOR_CRT_ENABLED }
+   * Value {@value}.
+   */
+  public static final boolean ANALYTICS_ACCELERATOR_CRT_ENABLED_DEFAULT = true;
+
 }