HADOOP-18993

Author: agilelab-tmnd1991

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

@@ -1360,4 +1360,22 @@ private Constants() {
    * Value: {@value}.
    */
   public static final boolean OPTIMIZED_COPY_FROM_LOCAL_DEFAULT = true;
+
+
+  /**
+   * To use the same classloader that loaded S3AFileSystem to al
+   * load the user extensions, such as {{fs.s3a.aws.credentials.provider}}.
+   * It is useful to turn this off for Apache Spark applications that 
+   * might load S3AFileSystem from the Spark distribution (Launcher classloader) 
+   * while users might want to provide custom extensions (loaded by Spark MutableClassloader).
+   * Default value: true.
+   */
+  public static final String AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER = 
+      "fs.s3a.extensions.isolated.classloader";
+
+
+  /**
+   * Default value for {{fs.s3a.extensions.isolated.classloader}}.
+   */
+  public static final boolean DEFAULT_AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER = true;
 }

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

@@ -534,7 +534,7 @@ public void initialize(URI name, Configuration originalConf)
       // fix up the classloader of the configuration to be whatever
       // classloader loaded this filesystem.
       // See: HADOOP-17372
-      conf.setClassLoader(this.getClass().getClassLoader());
+      isolateClassloader(conf, this.getClass().getClassLoader());
 
       // patch the Hadoop security providers
       patchSecurityCredentialProviders(conf);
@@ -718,6 +718,18 @@ public void initialize(URI name, Configuration originalConf)
     }
   }
 
+  void isolateClassloader(Configuration conf, ClassLoader classLoader) {
+    if (conf.getBoolean(Constants.AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER,
+        Constants.DEFAULT_AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER)) {
+      LOG.debug("Configuration classloader set to S3AFileSystem classloader: {}", classLoader);
+      conf.setClassLoader(classLoader);
+    } else {
+      LOG.debug("Configuration classloader not changed, support classes needed will be loaded " +
+          "from the classloader that instantiated the Configuration object: {}",
+          conf.getClassLoader());
+    }
+  }
+
   /**
    * Populates the configurations related to vectored IO operation
    * in the context which has to passed down to input streams.

hadoop-tools/hadoop-aws/src/site/markdown/tools/hadoop-aws/index.md

@@ -561,6 +561,30 @@ obtain the credentials needed to access AWS services in the role the EC2 VM
 was deployed as.
 This AWS credential provider is enabled in S3A by default.
 
+## Custom AWS Credential Providers and Apache Spark
+
+Apache Spark employs two class loaders, one that loads "distribution" (Spark + Hadoop) classes and one that
+loads custom user classes. If the user wants to load custom implementations of AWS Credential Providers
+through user provided jars will need to set the following configuration:
+
+```xml
+<property>
+  <name>fs.s3a.extensions.isolated.classloader</name>
+  <value>false</value>
+</property>
+<property>
+  <name>fs.s3a.aws.credentials.provider</name>
+  <value>CustomCredentialsProvider</value>
+</property>
+
+```
+
+If the following property is not set or set to true, the following exception will be thrown:
+
+```
+java.io.IOException: From option fs.s3a.aws.credentials.provider java.lang.ClassNotFoundException: Class CustomCredentialsProvider not found
+```
+
 
 ## <a name="hadoop_credential_providers"></a>Storing secrets with Hadoop Credential Providers
 

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

@@ -0,0 +1,43 @@
+package org.apache.hadoop.fs.s3a;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.junit.Assert;
+import org.junit.Test;
+
+public class TestS3AFileSystemIsolatedClassloader {
+
+  public static class CustomClassLoader extends ClassLoader {
+  }
+
+  private final ClassLoader classLoader = new CustomClassLoader();
+
+  @Test
+  public void isolatedClasspath() throws IOException {
+    try (S3AFileSystem fs = new S3AFileSystem()) {
+      Configuration conf = new Configuration();
+      conf.setBoolean(Constants.AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER, true);
+      fs.isolateClassloader(conf, classLoader);
+      Assert.assertTrue(conf.getClassLoader() instanceof CustomClassLoader);
+    }
+  }
+
+  @Test
+  public void isolatedClasspathDefault() throws IOException {
+    try (S3AFileSystem fs = new S3AFileSystem()) {
+      Configuration conf = new Configuration();
+      fs.isolateClassloader(conf, classLoader);
+      Assert.assertTrue(conf.getClassLoader() instanceof CustomClassLoader);
+    }
+  }
+
+  @Test
+  public void notIsolatedClasspath() throws IOException {
+    try (S3AFileSystem fs = new S3AFileSystem()) {
+      Configuration conf = new Configuration();
+      conf.setBoolean(Constants.AWS_S3_EXTENSIONS_ISOLATED_CLASSLOADER, false);
+      fs.isolateClassloader(conf, classLoader);
+      Assert.assertFalse(conf.getClassLoader() instanceof CustomClassLoader);
+    }
+  }
+}