Add support of hadoop-aws s3a SimpleAWSCredentialsProvider to S3DynamoDBLogStore

GrigorievNick · vkorukanti · commit 91f0650578dd · 2022-07-19T07:43:58.000-07:00
(Cherry-pick of 86cf73f) - Add support for `SimpleAWSCredentialsProvider` or `TemporaryAWSCredentialsProvider` in `spark.io.delta.storage.S3DynamoDBLogStore.credentials.provider` options. - Because delta rely on Spark and Hadoop FS storage layer, so it's obvious to have ability authorize in dynamo db client in same way as we authorize for s3. Resolves #1235. We use it in production with spark 3.2 on YARN 2.9.1 and my own fork of delta 1.2.1. Fork made from latest 1.2.1 with cherypicked multipart checkpoint commit. Scala 2.12 I have more than 100 tables, where data ingested every 10 minutes and multiple job work daily. Like retention and Row Level Update in some files. No. Except may be that [official example](https://docs.delta.io/latest/delta-storage.html#quickstart-s3-multi-cluster ) will work in any environment, and not only environment when Node where Spark App scheduled have configured AWS credentials. Please find more details about reason in #1235. Closes #1253 Signed-off-by: Venki Korukanti <venki.korukanti@gmail.com> GitOrigin-RevId: cbcc087457971c91d9908ac44398492bfa49d811
diff --git a/storage-s3-dynamodb/src/main/java/io/delta/storage/S3DynamoDBLogStore.java b/storage-s3-dynamodb/src/main/java/io/delta/storage/S3DynamoDBLogStore.java
@@ -16,6 +16,7 @@
 
 package io.delta.storage;
 
+import io.delta.storage.utils.ReflectionUtils;
 import org.apache.hadoop.fs.Path;
 
 import java.io.InterruptedIOException;
@@ -279,19 +280,12 @@ private void tryEnsureTableExists(Configuration hadoopConf) throws IOException {
 
     private AmazonDynamoDBClient getClient() throws java.io.IOException {
         try {
-            final AWSCredentialsProvider auth =
-                (AWSCredentialsProvider) Class.forName(credentialsProviderName)
-                    .getConstructor()
-                    .newInstance();
-            final AmazonDynamoDBClient client = new AmazonDynamoDBClient(auth);
+            final AWSCredentialsProvider awsCredentialsProvider =
+                    ReflectionUtils.createAwsCredentialsProvider(credentialsProviderName, initHadoopConf());
+            final AmazonDynamoDBClient client = new AmazonDynamoDBClient(awsCredentialsProvider);
             client.setRegion(Region.getRegion(Regions.fromName(regionName)));
             return client;
-        } catch (
-            ClassNotFoundException
-            | InstantiationException
-            | NoSuchMethodException
-            | IllegalAccessException
-            | java.lang.reflect.InvocationTargetException e) {
+        } catch (ReflectiveOperationException e) {
             throw new java.io.IOException(e);
         }
     }
diff --git a/storage-s3-dynamodb/src/main/java/io/delta/storage/utils/ReflectionUtils.java b/storage-s3-dynamodb/src/main/java/io/delta/storage/utils/ReflectionUtils.java
@@ -0,0 +1,59 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.storage.utils;
+
+import com.amazonaws.auth.AWSCredentialsProvider;
+import org.apache.hadoop.conf.Configuration;
+
+import java.util.Arrays;
+
+public class ReflectionUtils {
+
+    private static boolean readsCredsFromHadoopConf(Class<?> awsCredentialsProviderClass) {
+        return Arrays.stream(awsCredentialsProviderClass.getConstructors())
+                .anyMatch(constructor -> constructor.getParameterCount() == 1 &&
+                        Arrays.equals(constructor.getParameterTypes(), new Class[]{Configuration.class}));
+    }
+
+    /**
+     * Create AWS credentials provider from given provider classname and {@link Configuration}.
+     *
+     * It first check if AWS Credentials Provider class has constructor Hadoop configuration as parameter.
+     *   If yes - create instance of class using this constructor.
+     *   If no - create instance with empty parameters constructor.
+     *
+     * @param credentialsProviderClassName Fully qualified name of the desired credentials provider class.
+     * @param hadoopConf Hadoop configuration, used to create instance of AWS credentials
+     *                                      provider, if supported.
+     * @return {@link AWSCredentialsProvider} object, instantiated from the class @see {credentialsProviderClassName}
+     * @throws ReflectiveOperationException When AWS credentials provider constrictor do not matched.
+     *                                      Means class has neither an constructor with no args as input
+     *                                      nor constructor with only Hadoop configuration as argument.
+     */
+    public static AWSCredentialsProvider createAwsCredentialsProvider(
+            String credentialsProviderClassName,
+            Configuration hadoopConf) throws ReflectiveOperationException {
+        Class<?> awsCredentialsProviderClass = Class.forName(credentialsProviderClassName);
+        if (readsCredsFromHadoopConf(awsCredentialsProviderClass))
+            return (AWSCredentialsProvider) awsCredentialsProviderClass
+                    .getConstructor(Configuration.class)
+                    .newInstance(hadoopConf);
+        else
+            return (AWSCredentialsProvider) awsCredentialsProviderClass.getConstructor().newInstance();
+    }
+
+}
diff --git a/storage-s3-dynamodb/src/test/java/io/delta/storage/utils/ReflectionsUtilsSuiteHelper.java b/storage-s3-dynamodb/src/test/java/io/delta/storage/utils/ReflectionsUtilsSuiteHelper.java
@@ -0,0 +1,55 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.storage.utils;
+
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSCredentialsProvider;
+import org.apache.hadoop.conf.Configuration;
+
+public class ReflectionsUtilsSuiteHelper {
+    // this class only purpose to test DynamoDBLogStore logic to create AWS credentials provider with reflection.
+    public static class TestOnlyAWSCredentialsProviderWithHadoopConf implements AWSCredentialsProvider {
+
+        public TestOnlyAWSCredentialsProviderWithHadoopConf(Configuration hadoopConf) {}
+
+        @Override
+        public AWSCredentials getCredentials() {
+            return null;
+        }
+
+        @Override
+        public void refresh() {
+
+        }
+    }
+
+    // this class only purpose to test DynamoDBLogStore logic to create AWS credentials provider with reflection.
+    public static class TestOnlyAWSCredentialsProviderWithUnexpectedConstructor implements AWSCredentialsProvider {
+
+        public TestOnlyAWSCredentialsProviderWithUnexpectedConstructor(String hadoopConf) {}
+
+        @Override
+        public AWSCredentials getCredentials() {
+            return null;
+        }
+
+        @Override
+        public void refresh() {
+
+        }
+    }
+}
diff --git a/storage-s3-dynamodb/src/test/scala/io/delta/storage/utils/ReflectionsUtilsSuite.scala b/storage-s3-dynamodb/src/test/scala/io/delta/storage/utils/ReflectionsUtilsSuite.scala
@@ -0,0 +1,56 @@
+/*
+ * Copyright (2021) The Delta Lake Project Authors.
+ *
+ * Licensed 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 io.delta.storage.utils
+
+import com.amazonaws.auth.EnvironmentVariableCredentialsProvider
+import io.delta.storage.utils.ReflectionsUtilsSuiteHelper.TestOnlyAWSCredentialsProviderWithHadoopConf
+import org.apache.hadoop.conf.Configuration
+import org.scalatest.funsuite.AnyFunSuite
+
+class ReflectionsUtilsSuite extends AnyFunSuite {
+  private val emptyHadoopConf = new Configuration()
+
+  test("support AWS credentials provider with hadoop Configuration as constructor parameter") {
+    val awsProvider = ReflectionUtils.createAwsCredentialsProvider(
+      "io.delta.storage.utils.ReflectionsUtilsSuiteHelper" +
+        "$TestOnlyAWSCredentialsProviderWithHadoopConf",
+      emptyHadoopConf
+    )
+    assert(
+      awsProvider.isInstanceOf[TestOnlyAWSCredentialsProviderWithHadoopConf]
+    )
+  }
+
+  test("support AWS credentials provider with empty constructor(default from aws lib)") {
+    val awsProvider = ReflectionUtils.createAwsCredentialsProvider(
+      classOf[EnvironmentVariableCredentialsProvider].getCanonicalName,
+      emptyHadoopConf
+    )
+    assert(awsProvider.isInstanceOf[EnvironmentVariableCredentialsProvider])
+  }
+
+  test("do not support AWS credentials provider with unexpected constructors parameters") {
+    assertThrows[NoSuchMethodException] {
+      ReflectionUtils.createAwsCredentialsProvider(
+        "io.delta.storage.utils.ReflectionsUtilsSuiteHelper" +
+          "$TestOnlyAWSCredentialsProviderWithUnexpectedConstructor",
+        emptyHadoopConf
+      )
+    }
+  }
+
+}
