[AMORO-4063] Fix missing S3 AK/SK in Apache Paimon format (#4064)

Author: juntaozhang

amoro-common/src/main/java/org/apache/amoro/table/TableMetaStore.java

@@ -127,6 +127,7 @@ public class TableMetaStore implements Serializable {
   private final boolean disableAuth;
   private final String accessKey;
   private final String secretKey;
+  private final String storageType;
 
   private transient RuntimeContext runtimeContext;
   private transient String authInformation;
@@ -146,6 +147,7 @@ private TableMetaStore(
       String krbPrincipal,
       String accessKey,
       String secretKey,
+      String storageType,
       boolean disableAuth) {
     Preconditions.checkArgument(
         authMethod == null
@@ -165,6 +167,7 @@ private TableMetaStore(
     this.disableAuth = disableAuth;
     this.accessKey = accessKey;
     this.secretKey = secretKey;
+    this.storageType = storageType;
   }
 
   private TableMetaStore(Configuration configuration) {
@@ -179,6 +182,7 @@ private TableMetaStore(Configuration configuration) {
     this.krbPrincipal = null;
     this.accessKey = null;
     this.secretKey = null;
+    this.storageType = null;
     this.runtimeContext = new RuntimeContext();
     runtimeContext.setConfiguration(configuration);
   }
@@ -219,6 +223,10 @@ public String getSecretKey() {
     return secretKey;
   }
 
+  public String getStorageType() {
+    return storageType;
+  }
+
   public boolean isKerberosAuthMethod() {
     return AUTH_METHOD_KERBEROS.equalsIgnoreCase(authMethod);
   }
@@ -586,6 +594,7 @@ public static class Builder {
     private String krbPrincipal;
     private String accessKey;
     private String secretKey;
+    private String storageType;
     private boolean disableAuth = true;
     private final Map<String, String> properties = Maps.newHashMap();
     private Configuration configuration;
@@ -640,11 +649,12 @@ public Builder withBase64CoreSite(String encodedCoreSite) {
       return this;
     }
 
-    public Builder withAkSkAuth(String accessKey, String secretKey) {
+    public Builder withAkSkAuth(String accessKey, String secretKey, String storageType) {
       this.disableAuth = false;
       this.authMethod = AUTH_METHOD_AK_SK;
       this.accessKey = accessKey;
       this.secretKey = secretKey;
+      this.storageType = storageType;
       return this;
     }
 
@@ -785,6 +795,7 @@ public TableMetaStore build() {
         } else if (AUTH_METHOD_AK_SK.equals(authMethod)) {
           Preconditions.checkNotNull(accessKey);
           Preconditions.checkNotNull(secretKey);
+          Preconditions.checkNotNull(storageType);
         } else if (authMethod != null) {
           throw new IllegalArgumentException("Unsupported auth method:" + authMethod);
         }
@@ -805,6 +816,7 @@ public TableMetaStore build() {
             krbPrincipal,
             accessKey,
             secretKey,
+            storageType,
             disableAuth);
       }
     }

amoro-common/src/main/java/org/apache/amoro/utils/CatalogUtil.java

@@ -168,7 +168,9 @@ public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) {
               authType)) {
             String accessKey = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_ACCESS_KEY);
             String secretKey = authConfigs.get(CatalogMetaProperties.AUTH_CONFIGS_KEY_SECRET_KEY);
-            builder.withAkSkAuth(accessKey, secretKey);
+            String storageType =
+                catalogMeta.getStorageConfigs().get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE);
+            builder.withAkSkAuth(accessKey, secretKey, storageType);
           }
         }
       }
@@ -204,7 +206,9 @@ public static TableMetaStore buildMetaStore(CatalogMeta catalogMeta) {
               catalogMeta
                   .getCatalogProperties()
                   .get(CatalogMetaProperties.AUTH_CONFIGS_KEY_SECRET_KEY);
-          builder.withAkSkAuth(accessKey, secretKey);
+          String storageType =
+              catalogMeta.getStorageConfigs().get(CatalogMetaProperties.STORAGE_CONFIGS_KEY_TYPE);
+          builder.withAkSkAuth(accessKey, secretKey, storageType);
         }
       }
     }

amoro-format-paimon/src/main/java/org/apache/amoro/formats/paimon/PaimonCatalogFactory.java

@@ -63,7 +63,8 @@ public PaimonCatalog create(
                 HiveCatalogOptions.HIVE_CONF_DIR.key(), new File(url.getPath()).getParent()));
     if (CatalogMetaProperties.AUTH_CONFIGS_VALUE_TYPE_AK_SK.equalsIgnoreCase(
         metaStore.getAuthMethod())) {
-      if (CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_S3.equals(metastoreType)) {
+      if (CatalogMetaProperties.STORAGE_CONFIGS_VALUE_TYPE_S3.equalsIgnoreCase(
+          metaStore.getStorageType())) {
         // s3.access-key, s3.secret-key
         catalogProperties.put(PAIMON_S3_ACCESS_KEY, metaStore.getAccessKey());
         catalogProperties.put(PAIMON_S3_SECRET_KEY, metaStore.getSecretKey());