fix: abstracted-out

Author: arsalan-datazip

amoro-ams/src/main/java/org/apache/amoro/server/terminal/TerminalManager.java

@@ -42,8 +42,6 @@
 import org.apache.amoro.utils.CatalogUtil;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.iceberg.CatalogProperties;
-import org.apache.iceberg.aws.AwsClientProperties;
-import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -410,18 +408,11 @@ private void applyClientProperties(CatalogMeta catalogMeta) {
       // Glue client receives the same credentials that were provided for S3.
       if (CatalogMetaProperties.CATALOG_TYPE_GLUE.equalsIgnoreCase(catalogType)) {
         Map<String, String> props = catalogMeta.getCatalogProperties();
-        String accessKey = props.get(S3FileIOProperties.ACCESS_KEY_ID);
-        String secretKey = props.get(S3FileIOProperties.SECRET_ACCESS_KEY);
-        if (accessKey != null && secretKey != null) {
-          catalogMeta.putToCatalogProperties(
-              AwsClientProperties.CLIENT_CREDENTIALS_PROVIDER,
-              StaticAwsCredentialsProvider.class.getName());
-          catalogMeta.putToCatalogProperties(
-              "client.credentials-provider." + StaticAwsCredentialsProvider.ACCESS_KEY_ID,
-              accessKey);
-          catalogMeta.putToCatalogProperties(
-              "client.credentials-provider." + StaticAwsCredentialsProvider.SECRET_ACCESS_KEY,
-              secretKey);
+        Map<String, String> enriched = StaticAwsCredentialsProvider.applyGlueCredentials(props);
+        // Only merge when a new map was produced: same instance means nothing to add, and merging
+        // would re-iterate the live CatalogMeta map while mutating it.
+        if (enriched != null && enriched != props) {
+          enriched.forEach(catalogMeta::putToCatalogProperties);
         }
       }
     } else if (formats.contains(TableFormat.PAIMON) && "hive".equals(catalogType)) {

amoro-format-iceberg/src/main/java/org/apache/amoro/aws/StaticAwsCredentialsProvider.java

@@ -20,10 +20,13 @@
 
 package org.apache.amoro.aws;
 
+import org.apache.iceberg.aws.AwsClientProperties;
+import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 
+import java.util.HashMap;
 import java.util.Map;
 
 /**
@@ -53,6 +56,31 @@ public static AwsCredentialsProvider create(Map<String, String> properties) {
     return new StaticAwsCredentialsProvider(accessKeyId, secretAccessKey);
   }
 
+  /**
+   * Returns a copy of {@code properties} enriched with {@link
+   * AwsClientProperties#CLIENT_CREDENTIALS_PROVIDER} pointing at this class, so that Glue (and any
+   * other AWS client) uses the S3 AK/SK instead of the default credential chain. If {@code
+   * properties} is null, returns null. If either S3 key is absent, returns {@code properties}
+   * unchanged (same instance).
+   */
+  public static Map<String, String> applyGlueCredentials(Map<String, String> properties) {
+    if (properties == null) {
+      return null;
+    }
+    String accessKey = properties.get(S3FileIOProperties.ACCESS_KEY_ID);
+    String secretKey = properties.get(S3FileIOProperties.SECRET_ACCESS_KEY);
+    if (accessKey == null || secretKey == null) {
+      return properties;
+    }
+    Map<String, String> newProperties = new HashMap<>(properties);
+    newProperties.put(
+        AwsClientProperties.CLIENT_CREDENTIALS_PROVIDER,
+        StaticAwsCredentialsProvider.class.getName());
+    newProperties.put("client.credentials-provider." + ACCESS_KEY_ID, accessKey);
+    newProperties.put("client.credentials-provider." + SECRET_ACCESS_KEY, secretKey);
+    return newProperties;
+  }
+
   @Override
   public AwsCredentials resolveCredentials() {
     return credentials;

amoro-format-iceberg/src/main/java/org/apache/amoro/formats/iceberg/IcebergCatalogFactory.java

@@ -26,13 +26,10 @@
 import org.apache.amoro.aws.StaticAwsCredentialsProvider;
 import org.apache.amoro.properties.CatalogMetaProperties;
 import org.apache.amoro.shade.guava32.com.google.common.base.Preconditions;
-import org.apache.amoro.shade.guava32.com.google.common.collect.Maps;
 import org.apache.amoro.table.TableMetaStore;
 import org.apache.amoro.utils.MixedFormatCatalogUtil;
 import org.apache.commons.lang3.StringUtils;
 import org.apache.iceberg.CatalogUtil;
-import org.apache.iceberg.aws.AwsClientProperties;
-import org.apache.iceberg.aws.s3.S3FileIOProperties;
 import org.apache.iceberg.catalog.Catalog;
 
 import java.util.Map;
@@ -47,23 +44,8 @@ public FormatCatalog create(
         MixedFormatCatalogUtil.withIcebergCatalogInitializeProperties(
             name, metastoreType, properties);
 
-    // Setting the StaticAwsCredentialsProvider for Glue catalog to read the S3 credentials
     if (CatalogMetaProperties.CATALOG_TYPE_GLUE.equalsIgnoreCase(metastoreType)) {
-      Map<String, String> newProperties = Maps.newHashMap(properties);
-      String accessKey = newProperties.get(S3FileIOProperties.ACCESS_KEY_ID);
-      String secretKey = newProperties.get(S3FileIOProperties.SECRET_ACCESS_KEY);
-
-      if (accessKey != null && secretKey != null) {
-        newProperties.put(
-            AwsClientProperties.CLIENT_CREDENTIALS_PROVIDER,
-            StaticAwsCredentialsProvider.class.getName());
-        newProperties.put(
-            "client.credentials-provider." + StaticAwsCredentialsProvider.ACCESS_KEY_ID, accessKey);
-        newProperties.put(
-            "client.credentials-provider." + StaticAwsCredentialsProvider.SECRET_ACCESS_KEY,
-            secretKey);
-        properties = newProperties;
-      }
+      properties = StaticAwsCredentialsProvider.applyGlueCredentials(properties);
     }
 
     Catalog icebergCatalog =