HADOOP-18073. Upgrade AWS SDK to v2 in S3A

See aws_sdk_v2_changelog.md for details.

Co-authored-by: Ahmar Suhail <ahmarsu@amazon.co.uk>
Co-authored-by: Alessandro Passaro <alexpax@amazon.co.uk>

HADOOP-18073. Address review comments. (apache#31)

addresses review comments + yetus errors

Co-authored-by: Ahmar Suhail <ahmarsu@amazon.co.uk>

Move MultiObjectDeleteException to impl

Reinstate old constants

Move TransferManager initialization to ClientFactory

Add unit tests for BlockingEnumeration

Add unit tests for SelectEventStreamPublisher

updates new providers in TestS3AAWSCredentialsProvider to V2

update GET range referrer header logic to V2

adds in unit check for bytes

HADOOP-18565. Complete outstanding items for the AWS SDK V2 upgrade. (apache#5421)

Changes include
* use bundled transfer manager
* adds transfer listener to upload
* adds support for custom signers
* don't set default endpoint
* removes v1 sdk bundle, only use core package
* implements region caching
+ many more

Note: spotbugs is warning about inconsistent
synchronization in accessing a new s3a FS field.
This will be fixed in a follow-up patch.

Contributed by Ahmar Suhail

Author: ahmarsuhail

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

@@ -407,6 +407,10 @@ public final class StoreStatisticNames {
   public static final String MULTIPART_UPLOAD_LIST
       = "multipart_upload_list";
 
+  /** Probe for store region: {@value}. */
+  public static final String STORE_REGION_PROBE
+      = "store_region_probe";
+
   private StoreStatisticNames() {
   }
 

hadoop-project/pom.xml

@@ -184,6 +184,8 @@
     <surefire.fork.timeout>900</surefire.fork.timeout>
     <aws-java-sdk.version>1.12.499</aws-java-sdk.version>
     <hsqldb.version>2.7.1</hsqldb.version>
+    <aws-java-sdk-v2.version>2.19.12</aws-java-sdk-v2.version>
+    <awscrt.version>0.21.0</awscrt.version>
     <frontend-maven-plugin.version>1.11.2</frontend-maven-plugin.version>
     <jasmine-maven-plugin.version>2.1</jasmine-maven-plugin.version>
     <phantomjs-maven-plugin.version>0.7</phantomjs-maven-plugin.version>
@@ -1128,15 +1130,25 @@
       </dependency>
       <dependency>
         <groupId>com.amazonaws</groupId>
-        <artifactId>aws-java-sdk-bundle</artifactId>
+        <artifactId>aws-java-sdk-core</artifactId>
         <version>${aws-java-sdk.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>software.amazon.awssdk</groupId>
+        <artifactId>bundle</artifactId>
+        <version>${aws-java-sdk-v2.version}</version>
         <exclusions>
           <exclusion>
             <groupId>io.netty</groupId>
             <artifactId>*</artifactId>
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>software.amazon.awssdk.crt</groupId>
+        <artifactId>aws-crt</artifactId>
+        <version>${awscrt.version}</version>
+      </dependency>
       <dependency>
         <groupId>org.apache.mina</groupId>
         <artifactId>mina-core</artifactId>

hadoop-tools/hadoop-aws/pom.xml

@@ -496,7 +496,17 @@
     </dependency>
     <dependency>
       <groupId>com.amazonaws</groupId>
-      <artifactId>aws-java-sdk-bundle</artifactId>
+      <artifactId>aws-java-sdk-core</artifactId>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk</groupId>
+      <artifactId>bundle</artifactId>
+      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.awssdk.crt</groupId>
+      <artifactId>aws-crt</artifactId>
       <scope>compile</scope>
     </dependency>
     <dependency>

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

@@ -18,7 +18,7 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.amazonaws.AmazonServiceException;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
 
 /**
  * A 400 "Bad Request" exception was received.
@@ -36,7 +36,7 @@ public class AWSBadRequestException extends AWSServiceIOException {
    * @param cause the underlying cause
    */
   public AWSBadRequestException(String operation,
-      AmazonServiceException cause) {
+      AwsServiceException cause) {
     super(operation, cause);
   }
 }

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

@@ -18,34 +18,33 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.amazonaws.AmazonClientException;
-import com.amazonaws.SdkBaseException;
+import software.amazon.awssdk.core.exception.SdkException;
 import org.apache.hadoop.util.Preconditions;
 
 import java.io.IOException;
 
+
 /**
- * IOException equivalent of an {@link AmazonClientException}.
+ * IOException equivalent of an {@link SdkException}.
  */
 public class AWSClientIOException extends IOException {
 
   private final String operation;
 
   public AWSClientIOException(String operation,
-      SdkBaseException cause) {
+      SdkException cause) {
     super(cause);
     Preconditions.checkArgument(operation != null, "Null 'operation' argument");
     Preconditions.checkArgument(cause != null, "Null 'cause' argument");
     this.operation = operation;
   }
 
-  public AmazonClientException getCause() {
-    return (AmazonClientException) super.getCause();
+  public SdkException getCause() {
+    return (SdkException) super.getCause();
   }
 
   @Override
   public String getMessage() {
     return operation + ": " + getCause().getMessage();
   }
-
 }

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

@@ -27,12 +27,14 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-import com.amazonaws.AmazonClientException;
 import com.amazonaws.auth.AWSCredentials;
 import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.auth.AnonymousAWSCredentials;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.auth.BasicSessionCredentials;
 import org.apache.hadoop.classification.VisibleForTesting;
+import org.apache.hadoop.fs.s3a.adapter.V1V2AwsCredentialProviderAdapter;
 import org.apache.hadoop.util.Preconditions;
+
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,6 +45,12 @@
 import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.apache.hadoop.io.IOUtils;
 
+import software.amazon.awssdk.auth.credentials.AnonymousCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.core.exception.SdkException;
+
 /**
  * A list of providers.
  *
@@ -51,17 +59,17 @@
  * <ol>
  *   <li>Allows extra providers to be added dynamically.</li>
  *   <li>If any provider in the chain throws an exception other than
- *   an {@link AmazonClientException}, that is rethrown, rather than
+ *   an {@link SdkException}, that is rethrown, rather than
  *   swallowed.</li>
  *   <li>Has some more diagnostics.</li>
- *   <li>On failure, the last "relevant" AmazonClientException raised is
+ *   <li>On failure, the last "relevant" {@link SdkException} raised is
  *   rethrown; exceptions other than 'no credentials' have priority.</li>
- *   <li>Special handling of {@link AnonymousAWSCredentials}.</li>
+ *   <li>Special handling of {@link AnonymousCredentialsProvider}.</li>
  * </ol>
  */
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
-public final class AWSCredentialProviderList implements AWSCredentialsProvider,
+public final class AWSCredentialProviderList implements AwsCredentialsProvider,
     AutoCloseable {
 
   private static final Logger LOG = LoggerFactory.getLogger(
@@ -73,9 +81,9 @@ public final class AWSCredentialProviderList implements AWSCredentialsProvider,
       CREDENTIALS_REQUESTED_WHEN_CLOSED
       = "Credentials requested after provider list was closed";
 
-  private final List<AWSCredentialsProvider> providers = new ArrayList<>(1);
+  private final List<AwsCredentialsProvider> providers = new ArrayList<>(1);
   private boolean reuseLastProvider = true;
-  private AWSCredentialsProvider lastProvider;
+  private AwsCredentialsProvider lastProvider;
 
   private final AtomicInteger refCount = new AtomicInteger(1);
 
@@ -99,7 +107,9 @@ public AWSCredentialProviderList() {
    */
   public AWSCredentialProviderList(
       Collection<AWSCredentialsProvider> providers) {
-    this.providers.addAll(providers);
+    for (AWSCredentialsProvider provider: providers) {
+      this.providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
+    }
   }
 
   /**
@@ -110,6 +120,19 @@ public AWSCredentialProviderList(
   public AWSCredentialProviderList(final String name,
       final AWSCredentialsProvider... providerArgs) {
     setName(name);
+    for (AWSCredentialsProvider provider: providerArgs) {
+      this.providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
+    }
+  }
+
+  /**
+   * Create with an initial list of SDK V2 credential providers.
+   * @param name name for error messages, may be ""
+   * @param providerArgs provider list.
+   */
+  public AWSCredentialProviderList(final String name,
+      final AwsCredentialsProvider... providerArgs) {
+    setName(name);
     Collections.addAll(providers, providerArgs);
   }
 
@@ -127,12 +150,21 @@ public void setName(final String name) {
 
   /**
    * Add a new provider.
-   * @param p provider
+   * @param provider provider
    */
-  public void add(AWSCredentialsProvider p) {
-    providers.add(p);
+  public void add(AWSCredentialsProvider provider) {
+    providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
   }
 
+  /**
+   * Add a new SDK V2 provider.
+   * @param provider provider
+   */
+  public void add(AwsCredentialsProvider provider) {
+    providers.add(provider);
+  }
+
+
   /**
    * Add all providers from another list to this one.
    * @param other the other list.
@@ -142,15 +174,18 @@ public void addAll(AWSCredentialProviderList other) {
   }
 
   /**
-   * Refresh all child entries.
+   * This method will get credentials using SDK V2's resolveCredentials and then convert it into
+   * V1 credentials. This required by delegation token binding classes.
+   * @return SDK V1 credentials
    */
-  @Override
-  public void refresh() {
-    if (isClosed()) {
-      return;
-    }
-    for (AWSCredentialsProvider provider : providers) {
-      provider.refresh();
+  public AWSCredentials getCredentials() {
+    AwsCredentials credentials = resolveCredentials();
+    if (credentials instanceof AwsSessionCredentials) {
+      return new BasicSessionCredentials(credentials.accessKeyId(),
+          credentials.secretAccessKey(),
+          ((AwsSessionCredentials) credentials).sessionToken());
+    } else {
+      return new BasicAWSCredentials(credentials.accessKeyId(), credentials.secretAccessKey());
     }
   }
 
@@ -160,26 +195,26 @@ public void refresh() {
    * @return a set of credentials (possibly anonymous), for authenticating.
    */
   @Override
-  public AWSCredentials getCredentials() {
+  public AwsCredentials resolveCredentials() {
     if (isClosed()) {
       LOG.warn(CREDENTIALS_REQUESTED_WHEN_CLOSED);
       throw new NoAuthWithAWSException(name +
           CREDENTIALS_REQUESTED_WHEN_CLOSED);
     }
     checkNotEmpty();
     if (reuseLastProvider && lastProvider != null) {
-      return lastProvider.getCredentials();
+      return lastProvider.resolveCredentials();
     }
 
-    AmazonClientException lastException = null;
-    for (AWSCredentialsProvider provider : providers) {
+    SdkException lastException = null;
+    for (AwsCredentialsProvider provider : providers) {
       try {
-        AWSCredentials credentials = provider.getCredentials();
+        AwsCredentials credentials = provider.resolveCredentials();
         Preconditions.checkNotNull(credentials,
             "Null credentials returned by %s", provider);
-        if ((credentials.getAWSAccessKeyId() != null &&
-            credentials.getAWSSecretKey() != null)
-            || (credentials instanceof AnonymousAWSCredentials)) {
+        if ((credentials.accessKeyId() != null && credentials.secretAccessKey() != null) || (
+            provider instanceof AnonymousCredentialsProvider
+                || provider instanceof AnonymousAWSCredentialsProvider)) {
           lastProvider = provider;
           LOG.debug("Using credentials from {}", provider);
           return credentials;
@@ -196,7 +231,7 @@ public AWSCredentials getCredentials() {
         }
         LOG.debug("No credentials from {}: {}",
             provider, e.toString());
-      } catch (AmazonClientException e) {
+      } catch (SdkException e) {
         lastException = e;
         LOG.debug("No credentials provided by {}: {}",
             provider, e.toString(), e);
@@ -223,13 +258,13 @@ public AWSCredentials getCredentials() {
    * @return providers
    */
   @VisibleForTesting
-  List<AWSCredentialsProvider> getProviders() {
+  List<AwsCredentialsProvider> getProviders() {
     return providers;
   }
 
   /**
    * Verify that the provider list is not empty.
-   * @throws AmazonClientException if there are no providers.
+   * @throws SdkException if there are no providers.
    */
   public void checkNotEmpty() {
     if (providers.isEmpty()) {
@@ -317,7 +352,7 @@ public void close() {
     }
 
     // do this outside the synchronized block.
-    for (AWSCredentialsProvider p : providers) {
+    for (AwsCredentialsProvider p : providers) {
       if (p instanceof Closeable) {
         IOUtils.closeStream((Closeable) p);
       } else if (p instanceof AutoCloseable) {

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

@@ -18,14 +18,14 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.amazonaws.AmazonServiceException;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
 
 /**
  * Status code 443, no response from server. This is considered idempotent.
  */
 public class AWSNoResponseException extends AWSServiceIOException {
   public AWSNoResponseException(String operation,
-      AmazonServiceException cause) {
+      AwsServiceException cause) {
     super(operation, cause);
   }
 }

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

@@ -18,7 +18,7 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import com.amazonaws.AmazonServiceException;
+import software.amazon.awssdk.awscore.exception.AwsServiceException;
 
 /**
  * Request is redirected.
@@ -32,7 +32,7 @@ public class AWSRedirectException extends AWSServiceIOException {
    * @param cause the underlying cause
    */
   public AWSRedirectException(String operation,
-      AmazonServiceException cause) {
+      AwsServiceException cause) {
     super(operation, cause);
   }
 }