[#6024] feat(iceberg): refactor Iceberg credential code to reuse cred…

…ential component in Gravitino server (#6021)

### What changes were proposed in this pull request?

1. reuse `CatalogCredentialManager` to manage the credentials in Iceberg
catalog.
2. use `CatalogWrapperForREST` to manage some REST specific operations
like credential vending.
3. depracate `catalog-provider-type` , use `catalog-providers` instead.

### Why are the changes needed?

Fix: #6024 

### Does this PR introduce _any_ user-facing change?

yes, depracate `catalog-provider-type` , use `catalog-providers`, do
some compatibility work.

### How was this patch tested?
run with s3 token

catalogs/catalog-common/src/main/java/org/apache/gravitino/credential/CredentialConstants.java

@@ -20,7 +20,7 @@
 package org.apache.gravitino.credential;
 
 public class CredentialConstants {
-  public static final String CREDENTIAL_PROVIDER_TYPE = "credential-provider-type";
+  @Deprecated public static final String CREDENTIAL_PROVIDER_TYPE = "credential-provider-type";
   public static final String CREDENTIAL_PROVIDERS = "credential-providers";
   public static final String CREDENTIAL_CACHE_EXPIRE_RATIO = "credential-cache-expire-ratio";
   public static final String CREDENTIAL_CACHE_MAX_SIZE = "credential-cache-max-size";

clients/filesystem-hadoop3/src/test/java/org/apache/gravitino/filesystem/hadoop/integration/test/FilesetCatalogCredentialIT.java

@@ -86,9 +86,6 @@ public void startUp() throws Exception {
         S3TokenCredential.S3_TOKEN_CREDENTIAL_TYPE
             + ","
             + S3SecretKeyCredential.S3_SECRET_KEY_CREDENTIAL_TYPE);
-    properties.put(
-        CredentialConstants.CREDENTIAL_PROVIDER_TYPE,
-        S3SecretKeyCredential.S3_SECRET_KEY_CREDENTIAL_TYPE);
     properties.put(S3Properties.GRAVITINO_S3_ACCESS_KEY_ID, S3_ACCESS_KEY);
     properties.put(S3Properties.GRAVITINO_S3_SECRET_ACCESS_KEY, S3_SECRET_KEY);
     properties.put(S3Properties.GRAVITINO_S3_ENDPOINT, "s3.ap-southeast-2.amazonaws.com");

core/src/main/java/org/apache/gravitino/credential/CatalogCredentialManager.java

@@ -51,6 +51,17 @@ public Credential getCredential(String credentialType, CredentialContext context
     return credentialCache.getCredential(credentialCacheKey, cacheKey -> doGetCredential(cacheKey));
   }
 
+  // Get credential with only one credential provider.
+  public Credential getCredential(CredentialContext context) {
+    if (credentialProviders.size() == 0) {
+      throw new IllegalArgumentException("There are no credential provider for the catalog.");
+    } else if (credentialProviders.size() > 1) {
+      throw new UnsupportedOperationException(
+          "There are multiple credential providers for the catalog.");
+    }
+    return getCredential(credentialProviders.keySet().iterator().next(), context);
+  }
+
   @Override
   public void close() {
     credentialProviders
@@ -67,6 +78,11 @@ public void close() {
                     e);
               }
             });
+    try {
+      credentialCache.close();
+    } catch (IOException e) {
+      LOG.warn("Close credential cache failed, catalog: {}", catalogName, e);
+    }
   }
 
   private Credential doGetCredential(CredentialCacheKey credentialCacheKey) {

core/src/main/java/org/apache/gravitino/credential/CredentialUtils.java

@@ -19,30 +19,20 @@
 
 package org.apache.gravitino.credential;
 
-import com.google.common.base.Splitter;
-import com.google.common.collect.ImmutableSet;
 import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import java.util.Set;
 import java.util.function.Supplier;
 import java.util.stream.Collectors;
-import org.apache.gravitino.utils.PrincipalUtils;
+import org.apache.gravitino.credential.config.CredentialConfig;
 
 public class CredentialUtils {
 
-  private static final Splitter splitter = Splitter.on(",");
-
-  public static Credential vendCredential(CredentialProvider credentialProvider, String[] path) {
-    PathBasedCredentialContext pathBasedCredentialContext =
-        new PathBasedCredentialContext(
-            PrincipalUtils.getCurrentUserName(), ImmutableSet.copyOf(path), Collections.emptySet());
-    return credentialProvider.getCredential(pathBasedCredentialContext);
-  }
-
   public static Map<String, CredentialProvider> loadCredentialProviders(
       Map<String, String> catalogProperties) {
-    Set<String> credentialProviders =
-        CredentialUtils.getCredentialProvidersByOrder(() -> catalogProperties);
+    CredentialConfig credentialConfig = new CredentialConfig(catalogProperties);
+    List<String> credentialProviders = credentialConfig.get(CredentialConfig.CREDENTIAL_PROVIDERS);
 
     return credentialProviders.stream()
         .collect(
@@ -80,14 +70,8 @@ private static Set<String> getCredentialProvidersFromProperties(Map<String, Stri
       return Collections.emptySet();
     }
 
-    String providers = properties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
-    if (providers == null) {
-      return Collections.emptySet();
-    }
-    return splitter
-        .trimResults()
-        .omitEmptyStrings()
-        .splitToStream(providers)
+    CredentialConfig credentialConfig = new CredentialConfig(properties);
+    return credentialConfig.get(CredentialConfig.CREDENTIAL_PROVIDERS).stream()
         .collect(Collectors.toSet());
   }
 }

core/src/main/java/org/apache/gravitino/credential/config/CredentialConfig.java

@@ -20,6 +20,8 @@
 package org.apache.gravitino.credential.config;
 
 import com.google.common.collect.ImmutableMap;
+import java.util.Collections;
+import java.util.List;
 import java.util.Map;
 import org.apache.gravitino.Config;
 import org.apache.gravitino.config.ConfigBuilder;
@@ -67,6 +69,14 @@ public class CredentialConfig extends Config {
                   false /* reserved */))
           .build();
 
+  public static final ConfigEntry<List<String>> CREDENTIAL_PROVIDERS =
+      new ConfigBuilder(CredentialConstants.CREDENTIAL_PROVIDERS)
+          .doc("Credential providers, separated by comma.")
+          .version(ConfigConstants.VERSION_0_8_0)
+          .stringConf()
+          .toSequence()
+          .createWithDefault(Collections.emptyList());
+
   public static final ConfigEntry<Double> CREDENTIAL_CACHE_EXPIRE_RATIO =
       new ConfigBuilder(CredentialConstants.CREDENTIAL_CACHE_EXPIRE_RATIO)
           .doc(

dev/docker/iceberg-rest-server/rewrite_config.py

@@ -22,7 +22,8 @@
   "GRAVITINO_IO_IMPL" : "io-impl",
   "GRAVITINO_URI" : "uri",
   "GRAVITINO_WAREHOUSE" : "warehouse",
-  "GRAVITINO_CREDENTIAL_PROVIDER_TYPE" : "credential-provider-type",
+  "GRAVITINO_CREDENTIAL_PROVIDER_TYPE" : "credential-providers",
+  "GRAVITINO_CREDENTIAL_PROVIDERS" : "credential-providers",
   "GRAVITINO_GCS_CREDENTIAL_FILE_PATH" : "gcs-credential-file-path",
   "GRAVITINO_S3_ACCESS_KEY" : "s3-access-key-id",
   "GRAVITINO_S3_SECRET_KEY" : "s3-secret-access-key",