changes as per review comments

Author: ahmarsuhail

hadoop-tools/hadoop-aws/pom.xml

@@ -511,7 +511,7 @@
     <dependency>
       <groupId>software.amazon.encryption.s3</groupId>
       <artifactId>amazon-s3-encryption-client-java</artifactId>
-      <scope>compile</scope>
+      <scope>provided</scope>
     </dependency>
     <dependency>
       <groupId>software.amazon.eventstream</groupId>

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

@@ -96,37 +96,6 @@ public S3Client createS3Client(
         .build();
   }
 
-  @Override
-  public S3Client createS3EncryptionClient(final S3AsyncClient s3AsyncClient,
-      final S3Client s3Client, final CSEMaterials cseMaterials) {
-
-    S3EncryptionClient.Builder s3EncryptionClientBuilder =
-        S3EncryptionClient.builder().wrappedAsyncClient(s3AsyncClient).wrappedClient(s3Client)
-            .enableLegacyUnauthenticatedModes(true);
-
-    if (cseMaterials.getCseKeyType().equals(CSEMaterials.CSEKeyType.KMS)) {
-      s3EncryptionClientBuilder.kmsKeyId(cseMaterials.getKmsKeyId());
-    }
-
-    return s3EncryptionClientBuilder.build();
-  }
-
-
-  @Override
-  public S3AsyncClient createS3AsyncEncryptionClient(final S3AsyncClient s3AsyncClient,
-      final CSEMaterials cseMaterials) {
-
-    S3AsyncEncryptionClient.Builder s3EncryptionAsyncClientBuilder =
-        S3AsyncEncryptionClient.builder().wrappedClient(s3AsyncClient)
-            .enableLegacyUnauthenticatedModes(true);
-
-    if (cseMaterials.getCseKeyType().equals(CSEMaterials.CSEKeyType.KMS)) {
-      s3EncryptionAsyncClientBuilder.kmsKeyId(cseMaterials.getKmsKeyId());
-    }
-
-    return s3EncryptionAsyncClientBuilder.build();
-  }
-
   @Override
   public S3AsyncClient createS3AsyncClient(
       final URI uri,

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

@@ -0,0 +1,123 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.s3a;
+
+import java.io.IOException;
+import java.net.URI;
+
+import software.amazon.awssdk.services.s3.S3AsyncClient;
+import software.amazon.awssdk.services.s3.S3Client;
+import software.amazon.encryption.s3.S3AsyncEncryptionClient;
+import software.amazon.encryption.s3.S3EncryptionClient;
+
+import org.apache.hadoop.fs.s3a.impl.CSEMaterials;
+
+import static org.apache.hadoop.fs.s3a.impl.InstantiationIOException.unavailable;
+
+public class EncryptionS3ClientFactory extends DefaultS3ClientFactory {
+
+  private static final String ENCRYPTION_CLIENT_CLASSNAME =
+      "software.amazon.encryption.s3.S3EncryptionClient";
+
+  /**
+   * Encryption client availability.
+   */
+  private static final boolean ENCRYPTION_CLIENT_FOUND = checkForEncryptionClient();
+
+  /**
+   * S3Client to be wrapped by encryption client.
+   */
+  private S3Client s3Client;
+
+  /**
+   * S3AsyncClient to be wrapped by encryption client.
+   */
+  private S3AsyncClient s3AsyncClient;
+
+  private static boolean checkForEncryptionClient() {
+    try {
+      ClassLoader cl = EncryptionS3ClientFactory.class.getClassLoader();
+      cl.loadClass(ENCRYPTION_CLIENT_CLASSNAME);
+      LOG.debug("encryption client class {} found", ENCRYPTION_CLIENT_CLASSNAME);
+      return true;
+    } catch (Exception e) {
+      LOG.debug("encryption client class {} not found", ENCRYPTION_CLIENT_CLASSNAME, e);
+      return false;
+    }
+  }
+
+  /**
+   * Is the Encryption client available?
+   * @return true if it was found in the classloader
+   */
+  private static synchronized boolean isEncryptionClientAvailable() {
+    return ENCRYPTION_CLIENT_FOUND;
+  }
+
+
+  @Override
+  public S3Client createS3Client(URI uri, S3ClientCreationParameters parameters) throws IOException {
+
+    if (!isEncryptionClientAvailable()) {
+      throw unavailable(uri, ENCRYPTION_CLIENT_CLASSNAME, null, "No encryption client available");
+    }
+
+    s3Client = super.createS3Client(uri, parameters);
+    s3AsyncClient = super.createS3AsyncClient(uri, parameters);
+
+    return createS3EncryptionClient(parameters.getClientSideEncryptionMaterials());
+  }
+
+  @Override
+  public S3AsyncClient createS3AsyncClient(URI uri, S3ClientCreationParameters parameters) throws IOException {
+
+    if (!isEncryptionClientAvailable()) {
+      throw unavailable(uri, ENCRYPTION_CLIENT_CLASSNAME, null, "No encryption client available");
+    }
+
+    return createS3AsyncEncryptionClient(parameters.getClientSideEncryptionMaterials());
+  }
+
+  private S3Client createS3EncryptionClient(final CSEMaterials cseMaterials) {
+    S3EncryptionClient.Builder s3EncryptionClientBuilder =
+        S3EncryptionClient.builder().wrappedAsyncClient(s3AsyncClient).wrappedClient(s3Client)
+            .enableLegacyUnauthenticatedModes(true);
+
+    if (cseMaterials.getCseKeyType().equals(CSEMaterials.CSEKeyType.KMS)) {
+      s3EncryptionClientBuilder.kmsKeyId(cseMaterials.getKmsKeyId());
+    }
+
+    return s3EncryptionClientBuilder.build();
+  }
+
+
+  private S3AsyncClient createS3AsyncEncryptionClient(final CSEMaterials cseMaterials) {
+
+    S3AsyncEncryptionClient.Builder s3EncryptionAsyncClientBuilder =
+        S3AsyncEncryptionClient.builder().wrappedClient(s3AsyncClient)
+            .enableLegacyUnauthenticatedModes(true);
+
+    if (cseMaterials.getCseKeyType().equals(CSEMaterials.CSEKeyType.KMS)) {
+      s3EncryptionAsyncClientBuilder.kmsKeyId(cseMaterials.getKmsKeyId());
+    }
+
+    return s3EncryptionAsyncClientBuilder.build();
+  }
+
+}

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

@@ -97,7 +97,6 @@
 import software.amazon.awssdk.transfer.s3.model.CopyRequest;
 import software.amazon.awssdk.transfer.s3.model.FileUpload;
 import software.amazon.awssdk.transfer.s3.model.UploadFileRequest;
-import software.amazon.encryption.s3.materials.KmsKeyring;
 
 import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.fs.impl.prefetch.ExecutorServiceFuturePool;
@@ -982,6 +981,22 @@ private void bindAWSClient(URI name, boolean dtEnabled) throws IOException {
 
     Region region = getS3Region(configuredRegion);
 
+
+    S3ClientFactory clientFactory;
+    CSEMaterials cseMaterials = null;
+
+    if (isCSEEnabled) {
+      String kmsKeyId = getS3EncryptionKey(bucket, conf, true);
+
+      cseMaterials  = new CSEMaterials()
+          .withCSEKeyType(CSEMaterials.CSEKeyType.KMS)
+          .withKmsKeyId(kmsKeyId);
+
+      clientFactory = ReflectionUtils.newInstance(EncryptionS3ClientFactory.class, conf);
+    } else {
+      clientFactory = ReflectionUtils.newInstance(s3ClientFactoryClass, conf);
+    }
+
     S3ClientFactory.S3ClientCreationParameters parameters =
         new S3ClientFactory.S3ClientCreationParameters()
         .withCredentialSet(credentials)
@@ -997,24 +1012,12 @@ private void bindAWSClient(URI name, boolean dtEnabled) throws IOException {
         .withMultipartThreshold(multiPartThreshold)
         .withTransferManagerExecutor(unboundedThreadPool)
         .withRegion(region)
-        .withClientSideEncryptionEnabled(isCSEEnabled);
+        .withClientSideEncryptionEnabled(isCSEEnabled)
+        .withClientSideEncryptionMaterials(cseMaterials);
+
 
-    S3ClientFactory clientFactory = ReflectionUtils.newInstance(s3ClientFactoryClass, conf);
     s3Client = clientFactory.createS3Client(getUri(), parameters);
     createS3AsyncClient(clientFactory, parameters);
-
-    if (isCSEEnabled) {
-      String kmsKeyId = getS3EncryptionKey(bucket, conf, true);
-
-      CSEMaterials cseMaterials = new CSEMaterials()
-          .withCSEKeyType(CSEMaterials.CSEKeyType.KMS)
-          .withKmsKeyId(kmsKeyId);
-
-
-      s3Client = clientFactory.createS3EncryptionClient(s3AsyncClient, s3Client, cseMaterials);
-      s3AsyncClient = clientFactory.createS3AsyncEncryptionClient(s3AsyncClient, cseMaterials);
-    }
-
     transferManager =  clientFactory.createS3TransferManager(getS3AsyncClient());
   }
 

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

@@ -84,30 +84,6 @@ S3AsyncClient createS3AsyncClient(URI uri,
       S3ClientCreationParameters parameters) throws IOException;
 
 
-  /**
-   * Creates a new {@link software.amazon.encryption.s3.S3EncryptionClient}.
-   * Used when client side encryption is enabled.
-   *
-   * @param s3AsyncClient The asynchronous S3  client, will be used for cryptographic operations.
-   * @param s3Client The synchronous S3 client, will be used for non cryptographic operations.
-   * @param cseMaterials cse key and type to use
-   * @return S3EncryptionClient
-   */
-  S3Client createS3EncryptionClient(S3AsyncClient s3AsyncClient, S3Client s3Client,
-      CSEMaterials cseMaterials);
-
-
-  /**
-   * Creates a new {@link software.amazon.encryption.s3.S3AsyncEncryptionClient}.
-   * Used when client side encryption is enabled.
-   *
-   * @param s3AsyncClient The asynchronous S3  client, will be used for cryptographic operations.
-   * @param cseMaterials cse key and type to use
-   * @return S3AsyncEncryptionClient
-   */
-  S3AsyncClient createS3AsyncEncryptionClient(S3AsyncClient s3AsyncClient, CSEMaterials cseMaterials);
-
-
   /**
    * Creates a new {@link S3TransferManager}.
    *
@@ -202,6 +178,10 @@ final class S3ClientCreationParameters {
      */
     private Boolean isCSEEnabled;
 
+    /**
+     * Client side encryption materials.
+     */
+    private CSEMaterials cseMaterials;
 
     /**
      * List of execution interceptors to include in the chain
@@ -473,5 +453,24 @@ public S3ClientCreationParameters withClientSideEncryptionEnabled(final boolean
     public boolean isClientSideEncryptionEnabled() {
       return this.isCSEEnabled;
     }
+
+    /**
+     * Set the client side encryption materials.
+     *
+     * @param value new value
+     * @return the builder
+     */
+    public S3ClientCreationParameters withClientSideEncryptionMaterials(final CSEMaterials value) {
+      this.cseMaterials = value;
+      return this;
+    }
+
+    /**
+     * Get the client side encryption materials.
+     * @return client side encryption materials
+     */
+    public CSEMaterials getClientSideEncryptionMaterials() {
+      return this.cseMaterials;
+    }
   }
 }

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

@@ -365,7 +365,7 @@ public void shouldBeAbleToSwitchOnS3PathStyleAccessViaConfigProperty()
       throws Exception {
 
     conf = new Configuration();
-    unsetClientSideConfiguration(conf);
+    unsetEncryption(conf);
     conf.set(Constants.PATH_STYLE_ACCESS, Boolean.toString(true));
     assertTrue(conf.getBoolean(Constants.PATH_STYLE_ACCESS, false));
 
@@ -404,7 +404,7 @@ public void shouldBeAbleToSwitchOnS3PathStyleAccessViaConfigProperty()
   @Test
   public void testDefaultUserAgent() throws Exception {
     conf = new Configuration();
-    unsetClientSideConfiguration(conf);
+    unsetEncryption(conf);
     fs = S3ATestUtils.createTestFileSystem(conf);
     assertNotNull(fs);
     S3Client s3 = getS3Client("User Agent");
@@ -418,7 +418,7 @@ public void testDefaultUserAgent() throws Exception {
   @Test
   public void testCustomUserAgent() throws Exception {
     conf = new Configuration();
-    unsetClientSideConfiguration(conf);
+    unsetEncryption(conf);
     conf.set(Constants.USER_AGENT_PREFIX, "MyApp");
     fs = S3ATestUtils.createTestFileSystem(conf);
     assertNotNull(fs);
@@ -434,7 +434,7 @@ public void testCustomUserAgent() throws Exception {
   public void testRequestTimeout() throws Exception {
     conf = new Configuration();
     conf.set(REQUEST_TIMEOUT, "120");
-    unsetClientSideConfiguration(conf);
+    unsetEncryption(conf);
     fs = S3ATestUtils.createTestFileSystem(conf);
     S3Client s3 = getS3Client("Request timeout (ms)");
     SdkClientConfiguration clientConfiguration = getField(s3, SdkClientConfiguration.class,
@@ -579,7 +579,7 @@ public void testS3SpecificSignerOverride() throws Exception {
         .describedAs("Custom STS signer not called").isTrue();
   }
 
-  private void unsetClientSideConfiguration(Configuration conf) {
+  private void unsetEncryption(Configuration conf) {
     removeBaseAndBucketOverrides(conf, S3_ENCRYPTION_ALGORITHM);
     conf.set(Constants.S3_ENCRYPTION_ALGORITHM,
         S3AEncryptionMethods.NONE.getMethod());

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

@@ -63,20 +63,6 @@ public S3AsyncClient createS3AsyncClient(URI uri, final S3ClientCreationParamete
     return s3;
   }
 
-  @Override
-  public S3Client createS3EncryptionClient(final S3AsyncClient s3AsyncClient,
-      final S3Client s3Client, final CSEMaterials cseMaterials) {
-    S3Client s3 = mock(S3Client.class);
-    return s3;
-  }
-
-  @Override
-  public S3AsyncClient createS3AsyncEncryptionClient(final S3AsyncClient s3AsyncClient,
-      final CSEMaterials cseMaterials) {
-    S3AsyncClient s3 = mock(S3AsyncClient.class);
-    return s3;
-  }
-
 
   @Override
   public S3TransferManager createS3TransferManager(S3AsyncClient s3AsyncClient) {