[HUDI-6695] Use the AWS provider chain in Glue sync and add a new provider for STS assume role

hudi-aws/pom.xml

@@ -186,6 +186,12 @@
             <version>${aws.sdk.httpcore.version}</version>
         </dependency>
 
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>sts</artifactId>
+            <version>${aws.sdk.version}</version>
+        </dependency>
+
         <!-- Test -->
         <dependency>
             <groupId>org.apache.hudi</groupId>

hudi-aws/src/main/java/org/apache/hudi/aws/credentials/HoodieAWSCredentialsProviderFactory.java

@@ -36,6 +36,9 @@ public static AwsCredentialsProvider getAwsCredentialsProvider(Properties props)
 
   private static AwsCredentialsProvider getAwsCredentialsProviderChain(Properties props) {
     List<AwsCredentialsProvider> providers = new ArrayList<>();
+    if (HoodieConfigAWSAssumedRoleCredentialsProvider.validConf(props)) {
+      providers.add(new HoodieConfigAWSAssumedRoleCredentialsProvider(props));
+    }
     HoodieConfigAWSCredentialsProvider hoodieConfigAWSCredentialsProvider = new HoodieConfigAWSCredentialsProvider(props);
     if (hoodieConfigAWSCredentialsProvider.resolveCredentials() != null) {
       providers.add(hoodieConfigAWSCredentialsProvider);

hudi-aws/src/main/java/org/apache/hudi/aws/credentials/HoodieConfigAWSAssumedRoleCredentialsProvider.java

@@ -0,0 +1,65 @@
+/*
+ * 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.hudi.aws.credentials;
+
+import org.apache.hudi.common.util.StringUtils;
+import org.apache.hudi.config.HoodieAWSConfig;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+
+import java.util.Properties;
+
+/**
+ * Credentials provider which assumes AWS role from Hoodie config and fetches its credentials.
+ */
+public class HoodieConfigAWSAssumedRoleCredentialsProvider implements AwsCredentialsProvider {
+
+  private static final Logger LOG = LoggerFactory.getLogger(HoodieConfigAWSAssumedRoleCredentialsProvider.class);
+
+  private final StsAssumeRoleCredentialsProvider credentialsProvider;
+
+  public HoodieConfigAWSAssumedRoleCredentialsProvider(Properties props) {
+    String roleArn = props.getProperty(HoodieAWSConfig.AWS_ASSUME_ROLE_ARN.key());
+    AssumeRoleRequest req = AssumeRoleRequest.builder()
+          .roleArn(roleArn)
+          .roleSessionName("hoodie")
+          .build();
+    StsClient stsClient = StsClient.builder().build();
+
+    this.credentialsProvider = StsAssumeRoleCredentialsProvider.builder()
+          .stsClient(stsClient)
+          .refreshRequest(req)
+          .build();
+  }
+
+  public static boolean validConf(Properties props) {
+    String roleArn = props.getProperty(HoodieAWSConfig.AWS_ASSUME_ROLE_ARN.key());
+    return !StringUtils.isNullOrEmpty(roleArn);
+  }
+
+  @Override
+  public AwsCredentials resolveCredentials() {
+    return credentialsProvider.resolveCredentials();
+  }
+}

hudi-aws/src/main/java/org/apache/hudi/aws/sync/AWSGlueCatalogSyncClient.java

@@ -80,6 +80,7 @@
 import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_DATABASE_NAME;
 import static org.apache.hudi.sync.common.HoodieSyncConfig.META_SYNC_PARTITION_FIELDS;
 import static org.apache.hudi.sync.common.util.TableUtils.tableId;
+import org.apache.hudi.aws.credentials.HoodieAWSCredentialsProviderFactory;
 
 /**
  * This class implements all the AWS APIs to enable syncing of a Hudi Table with the
@@ -105,7 +106,9 @@ public class AWSGlueCatalogSyncClient extends HoodieSyncClient {
 
   public AWSGlueCatalogSyncClient(HiveSyncConfig config) {
     super(config);
-    this.awsGlue = GlueAsyncClient.builder().build();
+    this.awsGlue = GlueAsyncClient.builder()
+            .credentialsProvider(HoodieAWSCredentialsProviderFactory.getAwsCredentialsProvider(config.getProps()))
+            .build();
     this.databaseName = config.getStringOrDefault(META_SYNC_DATABASE_NAME);
     this.skipTableArchive = config.getBooleanOrDefault(GlueCatalogSyncClientConfig.GLUE_SKIP_TABLE_ARCHIVE);
     this.enableMetadataTable = Boolean.toString(config.getBoolean(GLUE_METADATA_FILE_LISTING)).toUpperCase();

hudi-aws/src/main/java/org/apache/hudi/config/HoodieAWSConfig.java

@@ -46,7 +46,7 @@
 @ConfigClassProperty(name = "Amazon Web Services Configs",
         groupName = ConfigGroups.Names.AWS,
         description = "Amazon Web Services configurations to access resources like Amazon DynamoDB (for locks),"
-            + " Amazon CloudWatch (metrics).")
+            + " Amazon CloudWatch (metrics) and Amazon Glue (metadata).")
 public class HoodieAWSConfig extends HoodieConfig {
   public static final ConfigProperty<String> AWS_ACCESS_KEY = ConfigProperty
       .key("hoodie.aws.access.key")
@@ -69,6 +69,13 @@ public class HoodieAWSConfig extends HoodieConfig {
       .sinceVersion("0.10.0")
       .withDocumentation("AWS session token");
 
+  public static final ConfigProperty<String> AWS_ASSUME_ROLE_ARN = ConfigProperty
+          .key("hoodie.aws.role.arn")
+          .noDefaultValue()
+          .markAdvanced()
+          .sinceVersion("0.13.2")
+          .withDocumentation("AWS Role ARN to assume");
+
   private HoodieAWSConfig() {
     super();
   }
@@ -89,6 +96,10 @@ public String getAWSSessionToken() {
     return getString(AWS_SESSION_TOKEN);
   }
 
+  public String getAWSAssumeRoleARN() {
+    return getString(AWS_ASSUME_ROLE_ARN);
+  }
+
   public static class Builder {
 
     private final HoodieAWSConfig awsConfig = new HoodieAWSConfig();
@@ -120,6 +131,11 @@ public HoodieAWSConfig.Builder withSessionToken(String sessionToken) {
       return this;
     }
 
+    public HoodieAWSConfig.Builder withAssumeRoleARN(String assumeRoleARN) {
+      awsConfig.setValue(AWS_ASSUME_ROLE_ARN, assumeRoleARN);
+      return this;
+    }
+
     public Builder withDynamoDBTable(String dynamoDbTableName) {
       awsConfig.setValue(DYNAMODB_LOCK_TABLE_NAME, dynamoDbTableName);
       return this;

hudi-aws/src/test/java/org/apache/hudi/aws/TestHoodieAWSCredentialsProviderFactory.java

@@ -39,4 +39,20 @@ public void testGetAWSCredentials() {
     assertEquals("random-secret-key", credentials.secretAccessKey());
     assertEquals("random-session-token", credentials.sessionToken());
   }
+
+  @Test
+  public void testGetAWSCredentialsWithInvalidAssumeRole() {
+    // This test is to ensure that the AWS credentials provider factory fallbacks to default credentials
+    // when the assume role ARN is invalid.
+    HoodieConfig cfg = new HoodieConfig();
+    cfg.setValue(HoodieAWSConfig.AWS_ACCESS_KEY, "random-access-key");
+    cfg.setValue(HoodieAWSConfig.AWS_SECRET_KEY, "random-secret-key");
+    cfg.setValue(HoodieAWSConfig.AWS_SESSION_TOKEN, "random-session-token");
+    cfg.setValue(HoodieAWSConfig.AWS_ASSUME_ROLE_ARN, "invalid-role-arn");
+    AwsSessionCredentials credentials = (AwsSessionCredentials) org.apache.hudi.aws.credentials.HoodieAWSCredentialsProviderFactory.getAwsCredentialsProvider(cfg.getProps()).resolveCredentials();
+    assertEquals("random-access-key", credentials.accessKeyId());
+    assertEquals("random-secret-key", credentials.secretAccessKey());
+    assertEquals("random-session-token", credentials.sessionToken());
+  }
+
 }