HADOOP-19733. S3A: load credentials providers using configured classloader

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

@@ -621,6 +621,9 @@ public static long dateToLong(final Date date) {
    * <li>a public default constructor.</li>
    * </ol>
    *
+   * Uses the Configuration's class loader, to respect the configured value of
+   * {@link Constants#AWS_S3_CLASSLOADER_ISOLATION}.
+   *
    * @param className name of class for which instance is to be created
    * @param conf configuration
    * @param uri URI of the FS
@@ -639,7 +642,8 @@ public static <InstanceT> InstanceT getInstanceFromReflection(String className,
       String methodName,
       String configKey) throws IOException {
     try {
-      Class<?> instanceClass = S3AUtils.class.getClassLoader().loadClass(className);
+      LOG.debug("Loading class {} with Configuration classloader {}", className, conf.getClassLoader());
+      Class<?> instanceClass = conf.getClassLoader().loadClass(className);
       if (Modifier.isAbstract(instanceClass.getModifiers())) {
         throw isAbstract(uri, className, configKey);
       }

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/ITestS3AFileSystemIsolatedClassloader.java

@@ -20,6 +20,7 @@
 
 import java.io.IOException;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.function.Consumer;
 
@@ -28,6 +29,14 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.s3a.impl.InstantiationIOException;
+
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.when;
 
 /**
  * Checks that classloader isolation for loading extension classes is applied
@@ -37,10 +46,33 @@
  */
 public class ITestS3AFileSystemIsolatedClassloader extends AbstractS3ATestBase {
 
+  private static String customClassName = "custom.class.name";
+
+  private static class CustomCredentialsProvider implements AwsCredentialsProvider {
+
+      public CustomCredentialsProvider() {
+      }
+
+      @Override
+      public AwsCredentials resolveCredentials() {
+          return null;
+      }
+
+  }
+
   private static class CustomClassLoader extends ClassLoader {
   }
 
-  private final ClassLoader customClassLoader = new CustomClassLoader();
+  private final ClassLoader customClassLoader = spy(new CustomClassLoader());
+  {
+    try {
+      doReturn(CustomCredentialsProvider.class)
+        .when(customClassLoader)
+        .loadClass(customClassName);
+    } catch (ClassNotFoundException ex) {
+      throw new RuntimeException(ex);
+    }
+  }
 
   private S3AFileSystem createNewTestFs(Configuration conf) throws IOException {
     S3AFileSystem fs = new S3AFileSystem();
@@ -77,19 +109,9 @@ private void assertInNewFilesystem(Map<String, String> confToSet, Consumer<FileS
     }
   }
 
-  private Map<String, String> mapOf() {
-    return new HashMap<>();
-  }
-
-  private Map<String, String> mapOf(String key, String value) {
-    HashMap<String, String> m = new HashMap<>();
-    m.put(key, value);
-    return m;
-  }
-
   @Test
   public void defaultIsolatedClassloader() throws IOException {
-    assertInNewFilesystem(mapOf(), (fs) -> {
+    assertInNewFilesystem(Map.of(), (fs) -> {
       Assertions.assertThat(fs.getConf().getClassLoader())
               .describedAs("The classloader used to load s3a fs extensions")
               .isNotEqualTo(Thread.currentThread().getContextClassLoader())
@@ -100,11 +122,26 @@ public void defaultIsolatedClassloader() throws IOException {
               .isEqualTo(fs.getClass().getClassLoader())
               .describedAs("the classloader that loaded the fs");
     });
+
+    Throwable thrown = Assertions.catchThrowable(() -> {
+      assertInNewFilesystem(
+              Map.of(Constants.AWS_CREDENTIALS_PROVIDER, customClassName),
+              (fs) -> {});
+    });
+
+    Assertions.assertThat(thrown)
+            .describedAs("thrown")
+            .isInstanceOf(InstantiationIOException.class);
+
+    Assertions.assertThat(thrown.getCause())
+            .describedAs("cause")
+            .isInstanceOf(ClassNotFoundException.class)
+            .hasMessageContaining(customClassName);
   }
 
   @Test
   public void isolatedClassloader() throws IOException {
-    assertInNewFilesystem(mapOf(Constants.AWS_S3_CLASSLOADER_ISOLATION, "true"), (fs) -> {
+    assertInNewFilesystem(Map.of(Constants.AWS_S3_CLASSLOADER_ISOLATION, "true"), (fs) -> {
       Assertions.assertThat(fs.getConf().getClassLoader())
               .describedAs("The classloader used to load s3a fs extensions")
               .isNotEqualTo(Thread.currentThread().getContextClassLoader())
@@ -115,11 +152,31 @@ public void isolatedClassloader() throws IOException {
               .isEqualTo(fs.getClass().getClassLoader())
               .describedAs("the classloader that loaded the fs");
     });
+
+    Throwable thrown = Assertions.catchThrowable(() -> {
+      assertInNewFilesystem(
+              Map.of(Constants.AWS_S3_CLASSLOADER_ISOLATION, "true",
+                     Constants.AWS_CREDENTIALS_PROVIDER, customClassName),
+              (fs) -> {});
+    });
+
+    Assertions.assertThat(thrown)
+            .describedAs("thrown")
+            .isInstanceOf(InstantiationIOException.class);
+
+    Assertions.assertThat(thrown.getCause())
+            .describedAs("cause")
+            .isInstanceOf(ClassNotFoundException.class)
+            .hasMessageContaining(customClassName);
   }
 
   @Test
   public void notIsolatedClassloader() throws IOException {
-    assertInNewFilesystem(mapOf(Constants.AWS_S3_CLASSLOADER_ISOLATION, "false"), (fs) -> {
+    Map<String, String> confToSet = Map.of(
+            Constants.AWS_S3_CLASSLOADER_ISOLATION, "false",
+            Constants.AWS_CREDENTIALS_PROVIDER, customClassName);
+
+    assertInNewFilesystem(confToSet, (fs) -> {
       Assertions.assertThat(fs.getConf().getClassLoader())
               .describedAs("The classloader used to load s3a fs extensions")
               .isEqualTo(Thread.currentThread().getContextClassLoader())
@@ -129,6 +186,16 @@ public void notIsolatedClassloader() throws IOException {
               .describedAs("The classloader used to load s3a fs extensions")
               .isNotEqualTo(fs.getClass().getClassLoader())
               .describedAs("the classloader that loaded the fs");
+
+      S3AFileSystem s3a = (S3AFileSystem) fs;
+      List<AwsCredentialsProvider> providers =
+              s3a.getS3AInternals().shareCredentials("test").getProviders();
+      Assertions.assertThat(providers)
+              .describedAs("providers")
+              .hasSize(1);
+      Assertions.assertThat(providers.get(0))
+              .describedAs("provider")
+              .isInstanceOf(CustomCredentialsProvider.class);
     });
   }
 }