HADOOP-18820. Cut AWS v1 support

LICENSE-binary

@@ -215,7 +215,6 @@ com.aliyun:aliyun-java-sdk-kms:2.11.0
 com.aliyun:aliyun-java-sdk-ram:3.1.0
 com.aliyun:aliyun-java-sdk-sts:3.0.0
 com.aliyun.oss:aliyun-sdk-oss:3.13.2
-com.amazonaws:aws-java-sdk-bundle:1.12.367
 com.cedarsoftware:java-util:1.9.0
 com.cedarsoftware:json-io:2.5.1
 com.fasterxml.jackson.core:jackson-annotations:2.12.7
@@ -364,6 +363,8 @@ org.objenesis:objenesis:2.6
 org.xerial.snappy:snappy-java:1.1.10.1
 org.yaml:snakeyaml:2.0
 org.wildfly.openssl:wildfly-openssl:1.1.3.Final
+software.amazon.awssdk:bundle:jar:2.19.12
+software.amazon.awssdk.crt:aws-crt:0.21.0
 
 
 --------------------------------------------------------------------------------

hadoop-project/pom.xml

@@ -185,6 +185,7 @@
     <aws-java-sdk.version>1.12.367</aws-java-sdk.version>
     <hsqldb.version>2.7.1</hsqldb.version>
     <aws-java-sdk-v2.version>2.19.12</aws-java-sdk-v2.version>
+    <aws.evenstream.version>1.0.1</aws.evenstream.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>
@@ -1132,18 +1133,29 @@
         <groupId>com.amazonaws</groupId>
         <artifactId>aws-java-sdk-core</artifactId>
         <version>${aws-java-sdk.version}</version>
+        <exclusions>
+          <exclusion>
+            <groupId>*</groupId>
+            <artifactId>*</artifactId>
+          </exclusion>
+        </exclusions>
       </dependency>
       <dependency>
         <groupId>software.amazon.awssdk</groupId>
         <artifactId>bundle</artifactId>
         <version>${aws-java-sdk-v2.version}</version>
         <exclusions>
           <exclusion>
-            <groupId>io.netty</groupId>
+            <groupId>*</groupId>
             <artifactId>*</artifactId>
           </exclusion>
         </exclusions>
       </dependency>
+      <dependency>
+        <groupId>software.amazon.eventstream</groupId>
+        <artifactId>eventstream</artifactId>
+        <version>${aws.evenstream.version}</version>
+      </dependency>
       <dependency>
         <groupId>software.amazon.awssdk.crt</groupId>
         <artifactId>aws-crt</artifactId>

hadoop-tools/hadoop-aws/pom.xml

@@ -464,6 +464,16 @@
                     <bannedImport>org.apache.hadoop.mapred.**</bannedImport>
                   </bannedImports>
                 </restrictImports>
+                <restrictImports>
+                  <includeTestCode>false</includeTestCode>
+                  <reason>Restrict AWS v1 imports to adapter code</reason>
+                  <exclusions>
+                    <exclusion>org.apache.hadoop.fs.s3a.adapter.V1ToV2AwsCredentialProviderAdapter</exclusion>
+                  </exclusions>
+                  <bannedImports>
+                    <bannedImport>com.amazonaws.**</bannedImport>
+                  </bannedImports>
+                </restrictImports>
               </rules>
             </configuration>
           </execution>
@@ -494,10 +504,14 @@
       <scope>test</scope>
       <type>test-jar</type>
     </dependency>
+
+    <!-- The v1 SDK is used at compilation time for adapter classes in
+         org.apache.hadoop.fs.s3a.adapter. It is not needed at runtime
+         unless a non-standard v1 credential provider is declared. -->
     <dependency>
       <groupId>com.amazonaws</groupId>
       <artifactId>aws-java-sdk-core</artifactId>
-      <scope>compile</scope>
+      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>software.amazon.awssdk</groupId>
@@ -507,7 +521,11 @@
     <dependency>
       <groupId>software.amazon.awssdk.crt</groupId>
       <artifactId>aws-crt</artifactId>
-      <scope>compile</scope>
+    </dependency>
+    <dependency>
+      <groupId>software.amazon.eventstream</groupId>
+      <artifactId>eventstream</artifactId>
+      <scope>test</scope>
     </dependency>
     <dependency>
       <groupId>org.assertj</groupId>

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

@@ -27,28 +27,21 @@
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 
-import com.amazonaws.auth.AWSCredentials;
-import com.amazonaws.auth.AWSCredentialsProvider;
-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;
 
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.classification.VisibleForTesting;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException;
 import org.apache.hadoop.fs.s3a.auth.NoAwsCredentialsException;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.util.Preconditions;
 
 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;
 
 /**
@@ -106,23 +99,8 @@ public AWSCredentialProviderList() {
    * @param providers provider list.
    */
   public AWSCredentialProviderList(
-      Collection<AWSCredentialsProvider> providers) {
-    for (AWSCredentialsProvider provider: providers) {
-      this.providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
-    }
-  }
-
-  /**
-   * Create with an initial list of providers.
-   * @param name name for error messages, may be ""
-   * @param providerArgs provider list.
-   */
-  public AWSCredentialProviderList(final String name,
-      final AWSCredentialsProvider... providerArgs) {
-    setName(name);
-    for (AWSCredentialsProvider provider: providerArgs) {
-      this.providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
-    }
+      Collection<AwsCredentialsProvider> providers) {
+    this.providers.addAll(providers);
   }
 
   /**
@@ -148,14 +126,6 @@ public void setName(final String name) {
     }
   }
 
-  /**
-   * Add a new provider.
-   * @param provider provider
-   */
-  public void add(AWSCredentialsProvider provider) {
-    providers.add(V1V2AwsCredentialProviderAdapter.adapt(provider));
-  }
-
   /**
    * Add a new SDK V2 provider.
    * @param provider provider
@@ -164,7 +134,6 @@ public void add(AwsCredentialsProvider provider) {
     providers.add(provider);
   }
 
-
   /**
    * Add all providers from another list to this one.
    * @param other the other list.
@@ -174,19 +143,11 @@ public void addAll(AWSCredentialProviderList other) {
   }
 
   /**
-   * 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
+   * Was an implementation of the v1 refresh; now just
+   * a no-op.
    */
-  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());
-    }
+  @Deprecated
+  public void refresh() {
   }
 
   /**
@@ -257,8 +218,7 @@ public AwsCredentials resolveCredentials() {
    *
    * @return providers
    */
-  @VisibleForTesting
-  List<AwsCredentialsProvider> getProviders() {
+  public List<AwsCredentialsProvider> getProviders() {
     return providers;
   }
 
@@ -290,9 +250,11 @@ public String listProviderNames() {
    */
   @Override
   public String toString() {
-    return "AWSCredentialProviderList[" +
-        name +
-        "refcount= " + refCount.get() + ": [" +
+    return "AWSCredentialProviderList"
+        + " name=" + name
+        + "; refcount= " + refCount.get()
+        + "; size="+ providers.size()
+        + ": [" +
         StringUtils.join(providers, ", ") + ']'
         + (lastProvider != null ? (" last provider: " + lastProvider) : "");
   }

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

@@ -65,7 +65,7 @@ private static class FailureInjectionInterceptor implements ExecutionInterceptor
      */
     private final AtomicLong failureCounter = new AtomicLong(0);
 
-   FailureInjectionInterceptor(FailureInjectionPolicy policy) {
+    FailureInjectionInterceptor(FailureInjectionPolicy policy) {
       this.policy = policy;
     }
 

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

@@ -75,6 +75,7 @@ public void bytesTransferred(TransferListener.Context.BytesTransferred context)
   /**
    * Method to invoke after upload has completed.
    * This can handle race conditions in setup/teardown.
+   * @param upload upload which has just completed.
    * @return the number of bytes which were transferred after the notification
    */
   public long uploadCompleted(ObjectTransfer upload) {

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

@@ -230,7 +230,7 @@
 import static org.apache.hadoop.fs.s3a.S3AUtils.*;
 import static org.apache.hadoop.fs.s3a.Statistic.*;
 import static org.apache.hadoop.fs.s3a.audit.S3AAuditConstants.INITIALIZE_SPAN;
-import static org.apache.hadoop.fs.s3a.auth.AwsCredentialListProvider.createAWSCredentialProviderSet;
+import static org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory.createAWSCredentialProviderList;
 import static org.apache.hadoop.fs.s3a.auth.RolePolicies.STATEMENT_ALLOW_SSE_KMS_RW;
 import static org.apache.hadoop.fs.s3a.auth.RolePolicies.allowS3Operations;
 import static org.apache.hadoop.fs.s3a.auth.delegation.S3ADelegationTokens.TokenIssuingPolicy.NoTokensAvailable;
@@ -852,7 +852,7 @@ protected void verifyBucketExists() throws UnknownStoreException, IOException {
 
     if(!trackDurationAndSpan(
         STORE_EXISTS_PROBE, bucket, null, () ->
-            invoker.retry("doestBucketExist", bucket, true, () -> {
+            invoker.retry("doesBucketExist", bucket, true, () -> {
               try {
                 if (BUCKET_REGIONS.containsKey(bucket)) {
                   return true;
@@ -952,7 +952,7 @@ private void bindAWSClient(URI name, boolean dtEnabled) throws IOException {
       uaSuffix = tokens.getUserAgentField();
     } else {
       // DT support is disabled, so create the normal credential chain
-      credentials = createAWSCredentialProviderSet(name, conf);
+      credentials = createAWSCredentialProviderList(name, conf);
     }
     LOG.debug("Using credential provider {}", credentials);
     Class<? extends S3ClientFactory> s3ClientFactoryClass = conf.getClass(

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

@@ -75,6 +75,10 @@
 import static org.apache.commons.lang3.StringUtils.isEmpty;
 import static org.apache.hadoop.fs.s3a.Constants.*;
 import static org.apache.hadoop.fs.s3a.impl.ErrorTranslation.isUnknownBucket;
+import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.instantiationException;
+import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.isAbstract;
+import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.isNotInstanceOf;
+import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.unsupportedConstructor;
 import static org.apache.hadoop.fs.s3a.impl.InternalConstants.*;
 import static org.apache.hadoop.io.IOUtils.cleanupWithLogger;
 import static org.apache.hadoop.util.functional.RemoteIterators.filteringRemoteIterator;
@@ -89,9 +93,6 @@
 public final class S3AUtils {
 
   private static final Logger LOG = LoggerFactory.getLogger(S3AUtils.class);
-  static final String CONSTRUCTOR_EXCEPTION = "constructor exception";
-  static final String INSTANTIATION_EXCEPTION
-      = "instantiation exception";
 
   static final String ENDPOINT_KEY = "Endpoint";
 
@@ -562,15 +563,65 @@ public static long dateToLong(final Date date) {
    * @return instance of the specified class
    * @throws IOException on any problem
    */
-  @SuppressWarnings("unchecked")
-  public static <InstanceT> InstanceT getInstanceFromReflection(Class<?> instanceClass,
-      Configuration conf, @Nullable URI uri, Class<?> interfaceImplemented, String methodName,
+  public static <InstanceT> InstanceT getInstanceFromReflection(
+      Class<?> instanceClass,
+      Configuration conf,
+      @Nullable URI uri,
+      Class<InstanceT> interfaceImplemented,
+      String methodName,
       String configKey) throws IOException {
 
-    String className = instanceClass.getName();
+    return getInstanceFromReflection(instanceClass.getName(),
+        conf,
+        uri,
+        interfaceImplemented,
+        methodName,
+        configKey);
+  }
 
+  /**
+   * Creates an instance of a class using reflection. The
+   * class must implement one of the following means of construction, which are
+   * attempted in order:
+   *
+   * <ol>
+   * <li>a public constructor accepting java.net.URI and
+   *     org.apache.hadoop.conf.Configuration</li>
+   * <li>a public constructor accepting
+   *    org.apache.hadoop.conf.Configuration</li>
+   * <li>a public static method named as per methodName, that accepts no
+   *    arguments and returns an instance of
+   *    specified type, or</li>
+   * <li>a public default constructor.</li>
+   * </ol>
+   *
+   * @param className name of class for which instance is to be created
+   * @param conf configuration
+   * @param uri URI of the FS
+   * @param interfaceImplemented interface that this class implements
+   * @param methodName name of factory method to be invoked
+   * @param configKey config key under which this class is specified
+   * @param <InstanceT> Instance of class
+   * @return instance of the specified class
+   * @throws IOException on any problem
+   */
+  @SuppressWarnings("unchecked")
+  public static <InstanceT> InstanceT getInstanceFromReflection(String className,
+      Configuration conf,
+      @Nullable URI uri,
+      Class<? extends InstanceT> interfaceImplemented,
+      String methodName,
+      String configKey) throws IOException {
     try {
-      Constructor cons = null;
+      Class<?> instanceClass = S3AUtils.class.getClassLoader().loadClass(className);
+      if (Modifier.isAbstract(instanceClass.getModifiers())) {
+        throw isAbstract(uri, className, configKey);
+      }
+      if (!interfaceImplemented.isAssignableFrom(instanceClass)) {
+        throw isNotInstanceOf(uri, className, interfaceImplemented.getName(), configKey);
+
+      }
+      Constructor cons;
       if (conf != null) {
         // new X(uri, conf)
         cons = getConstructor(instanceClass, URI.class, Configuration.class);
@@ -598,10 +649,7 @@ public static <InstanceT> InstanceT getInstanceFromReflection(Class<?> instanceC
       }
 
       // no supported constructor or factory method found
-      throw new IOException(String.format("%s " + CONSTRUCTOR_EXCEPTION
-          + ".  A class specified in %s must provide a public constructor "
-          + "of a supported signature, or a public factory method named "
-          + "create that accepts no arguments.", className, configKey));
+      throw unsupportedConstructor(uri, className, configKey);
     } catch (InvocationTargetException e) {
       Throwable targetException = e.getTargetException();
       if (targetException == null) {
@@ -613,12 +661,11 @@ public static <InstanceT> InstanceT getInstanceFromReflection(Class<?> instanceC
         throw translateException("Instantiate " + className, "", (SdkException) targetException);
       } else {
         // supported constructor or factory method found, but the call failed
-        throw new IOException(className + " " + INSTANTIATION_EXCEPTION + ": " + targetException,
-            targetException);
+        throw instantiationException(uri, className, configKey, targetException);
       }
     } catch (ReflectiveOperationException | IllegalArgumentException e) {
       // supported constructor or factory method found, but the call failed
-      throw new IOException(className + " " + INSTANTIATION_EXCEPTION + ": " + e, e);
+      throw instantiationException(uri, className, configKey, e);
     }
   }