[fix](glue) support glue on aws

fe/fe-core/src/main/java/com/amazonaws/glue/catalog/credentials/ConfigurationAWSCredentialsProvider2x.java

@@ -0,0 +1,50 @@
+// 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 com.amazonaws.glue.catalog.credentials;
+
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
+import java.util.Map;
+
+/**
+ * This is credentials provider for AWS SDK 2.x, comparing to ConfigurationAWSCredentialsProvider,
+ * which is for AWS SDK 1.x.
+ * See: https://docs.aws.amazon.com/sdk-for-java/latest/developer-guide/migration-client-credentials.html
+ */
+public class ConfigurationAWSCredentialsProvider2x implements AwsCredentialsProvider {
+
+    private AwsBasicCredentials awsBasicCredentials;
+
+    private ConfigurationAWSCredentialsProvider2x(AwsBasicCredentials awsBasicCredentials) {
+        this.awsBasicCredentials = awsBasicCredentials;
+    }
+
+    @Override
+    public AwsCredentials resolveCredentials() {
+        return awsBasicCredentials;
+    }
+
+    public static AwsCredentialsProvider create(Map<String, String> config) {
+        String ak = config.get("glue.access_key");
+        String sk = config.get("glue.secret_key");
+        AwsBasicCredentials awsBasicCredentials = AwsBasicCredentials.create(ak, sk);
+        return new ConfigurationAWSCredentialsProvider2x(awsBasicCredentials);
+    }
+}

fe/fe-core/src/main/java/org/apache/doris/datasource/property/PropertyConverter.java

@@ -520,6 +520,10 @@ private static Map<String, String> convertToGlueProperties(Map<String, String> p
             // glue ak sk for iceberg
             props.putIfAbsent(GlueProperties.ACCESS_KEY, credential.getAccessKey());
             props.putIfAbsent(GlueProperties.SECRET_KEY, credential.getSecretKey());
+            props.putIfAbsent(GlueProperties.CLIENT_CREDENTIALS_PROVIDER,
+                    "com.amazonaws.glue.catalog.credentials.ConfigurationAWSCredentialsProvider2x");
+            props.putIfAbsent(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_AK, credential.getAccessKey());
+            props.putIfAbsent(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_SK, credential.getSecretKey());
         }
         // set glue client metadata
         if (props.containsKey(GlueProperties.ENDPOINT)) {

fe/fe-core/src/main/java/org/apache/doris/datasource/property/constants/GlueProperties.java

@@ -32,9 +32,14 @@ public class GlueProperties extends BaseProperties {
     public static final String SECRET_KEY = "glue.secret_key";
     public static final String SESSION_TOKEN = "glue.session_token";
 
+    public static final String CLIENT_CREDENTIALS_PROVIDER = "client.credentials-provider";
+    public static final String CLIENT_CREDENTIALS_PROVIDER_AK = "client.credentials-provider.glue.access_key";
+    public static final String CLIENT_CREDENTIALS_PROVIDER_SK = "client.credentials-provider.glue.secret_key";
+
     public static final List<String> META_KEYS = Arrays.asList(AWSGlueConfig.AWS_GLUE_ENDPOINT,
             AWSGlueConfig.AWS_REGION, AWSGlueConfig.AWS_GLUE_ACCESS_KEY, AWSGlueConfig.AWS_GLUE_SECRET_KEY,
-            AWSGlueConfig.AWS_GLUE_SESSION_TOKEN);
+            AWSGlueConfig.AWS_GLUE_SESSION_TOKEN, CLIENT_CREDENTIALS_PROVIDER, CLIENT_CREDENTIALS_PROVIDER_AK,
+            CLIENT_CREDENTIALS_PROVIDER_SK);
 
     public static CloudCredential getCredential(Map<String, String> props) {
         return getCloudCredential(props, ACCESS_KEY, SECRET_KEY, SESSION_TOKEN);

fe/fe-core/src/test/java/org/apache/doris/datasource/property/PropertyConverterTest.java

@@ -34,8 +34,11 @@
 import org.apache.doris.common.Pair;
 import org.apache.doris.common.UserException;
 import org.apache.doris.common.jmockit.Deencapsulation;
+import org.apache.doris.common.util.PrintableMap;
+import org.apache.doris.datasource.ExternalCatalog;
 import org.apache.doris.datasource.hive.HMSExternalCatalog;
 import org.apache.doris.datasource.iceberg.IcebergExternalCatalog;
+import org.apache.doris.datasource.iceberg.IcebergGlueExternalCatalog;
 import org.apache.doris.datasource.maxcompute.MaxComputeExternalCatalog;
 import org.apache.doris.datasource.property.constants.CosProperties;
 import org.apache.doris.datasource.property.constants.DLFProperties;
@@ -777,4 +780,42 @@ public void testMetaPropertiesConvertor() {
         Assertions.assertEquals(9, res.size());
         Assertions.assertEquals("north", res.get(S3Properties.Env.REGION));
     }
+
+    @Test
+    public void testGluePropertiesConvertor() throws Exception {
+        Map<String, String> originProps = Maps.newHashMap();
+        originProps.put(GlueProperties.ACCESS_KEY, "ak");
+        originProps.put(GlueProperties.SECRET_KEY, "sk");
+        originProps.put(GlueProperties.ENDPOINT, "https://glue.us-east-1.amazonaws.com");
+        originProps.put("type", "iceberg");
+        originProps.put("iceberg.catalog.type", "glue");
+
+        Map<String, String> convertedProps = PropertyConverter.convertToMetaProperties(originProps);
+        System.out.println(convertedProps);
+        Assertions.assertEquals("com.amazonaws.glue.catalog.credentials.ConfigurationAWSCredentialsProvider2x",
+                convertedProps.get(GlueProperties.CLIENT_CREDENTIALS_PROVIDER));
+        Assertions.assertEquals("ak", convertedProps.get(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_AK));
+        Assertions.assertEquals("sk", convertedProps.get(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_SK));
+
+        String createIceGlue = "CREATE CATALOG iceglue PROPERTIES (\n"
+                + "    \"type\"=\"iceberg\",\n"
+                + "    \"iceberg.catalog.type\" = \"glue\",\n"
+                + "    \"glue.endpoint\" = \"https://glue.us-east-1.amazonaws.com/\",\n"
+                + "    \"glue.access_key\" = \"ak123\",\n"
+                + "    \"glue.secret_key\" = \"sk123\"\n"
+                + ");";
+        CreateCatalogStmt analyzedStmt = createStmt(createIceGlue);
+        IcebergExternalCatalog icebergExternalCatalog = createAndGetIcebergCatalog(analyzedStmt, "iceglue");
+        Assertions.assertTrue(icebergExternalCatalog instanceof IcebergGlueExternalCatalog);
+        IcebergGlueExternalCatalog glueCatalog = (IcebergGlueExternalCatalog) icebergExternalCatalog;
+
+        PrintableMap<String, String> printableMap = new PrintableMap<>(glueCatalog.getProperties(), "=", true, true,
+                true, true);
+        printableMap.setAdditionalHiddenKeys(ExternalCatalog.HIDDEN_PROPERTIES);
+        String result = printableMap.toString();
+        System.out.println(result);
+        Assertions.assertTrue(!result.contains(GlueProperties.CLIENT_CREDENTIALS_PROVIDER));
+        Assertions.assertTrue(!result.contains(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_AK));
+        Assertions.assertTrue(!result.contains(GlueProperties.CLIENT_CREDENTIALS_PROVIDER_SK));
+    }
 }