[#11263] feat(authz): Extend credential vending to Hive/Iceberg/Glue/JDBC catalog types

catalogs/catalog-jdbc-common/src/main/java/org/apache/gravitino/catalog/jdbc/JdbcCatalog.java

@@ -175,7 +175,9 @@ private Map<String, String> applyDefaultCredentialProviders(Map<String, String>
 
     String jdbcUser = properties.get(JdbcConfig.USERNAME.getKey());
     String jdbcPassword = properties.get(JdbcConfig.PASSWORD.getKey());
-    if (StringUtils.isNotBlank(jdbcUser) && StringUtils.isNotBlank(jdbcPassword)) {
+    // Register provider when user is configured; allow empty-string password (e.g. StarRocks
+    // default). Only skip when password is absent (null) entirely without a user.
+    if (StringUtils.isNotBlank(jdbcUser) && jdbcPassword != null) {
       properties.put(CredentialConstants.CREDENTIAL_PROVIDERS, JdbcCredential.JDBC_CREDENTIAL_TYPE);
     }
 

catalogs/catalog-jdbc-common/src/test/java/org/apache/gravitino/catalog/jdbc/TestJdbcCatalogCredential.java

@@ -109,6 +109,38 @@ void testJdbcCatalogDefaultCredentialProviders() {
     Assertions.assertEquals(JdbcCredential.JDBC_CREDENTIAL_TYPE, credentialProviders);
   }
 
+  @Test
+  void testJdbcCatalogDefaultCredentialProvidersWithEmptyPassword() {
+    AuditInfo auditInfo =
+        AuditInfo.builder().withCreator("creator").withCreateTime(Instant.now()).build();
+
+    // Empty-string password (e.g. StarRocks default) should still register the credential provider
+    Map<String, String> jdbcProps = Maps.newHashMap();
+    jdbcProps.put(JdbcConfig.JDBC_URL.getKey(), "jdbc:mysql://localhost:9030/");
+    jdbcProps.put(JdbcConfig.JDBC_DRIVER.getKey(), "com.mysql.cj.jdbc.Driver");
+    jdbcProps.put(JdbcConfig.USERNAME.getKey(), "root");
+    jdbcProps.put(JdbcConfig.PASSWORD.getKey(), "");
+
+    CatalogEntity jdbcEntity =
+        CatalogEntity.builder()
+            .withId(7L)
+            .withName("jdbc-catalog-empty-password")
+            .withNamespace(Namespace.of("metalake"))
+            .withType(TestableJdbcCatalog.Type.RELATIONAL)
+            .withProvider("jdbc-starrocks")
+            .withAuditInfo(auditInfo)
+            .withProperties(jdbcProps)
+            .build();
+
+    TestableJdbcCatalog jdbcCatalog = new TestableJdbcCatalog();
+    jdbcCatalog.withCatalogConf(jdbcProps).withCatalogEntity(jdbcEntity);
+    Map<String, String> properties = jdbcCatalog.propertiesWithCredentialProviders();
+
+    String credentialProviders = properties.get(CredentialConstants.CREDENTIAL_PROVIDERS);
+    Assertions.assertNotNull(credentialProviders);
+    Assertions.assertEquals(JdbcCredential.JDBC_CREDENTIAL_TYPE, credentialProviders);
+  }
+
   @Test
   void testJdbcCatalogNoCredentialProvidersWithoutPassword() {
     AuditInfo auditInfo =

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

@@ -53,7 +53,7 @@ public String credentialType() {
   @Nullable
   @Override
   public Credential getCredential(CredentialContext context) {
-    if (StringUtils.isBlank(jdbcUser) || StringUtils.isBlank(jdbcPassword)) {
+    if (StringUtils.isBlank(jdbcUser) || jdbcPassword == null) {
       return null;
     }
     return new JdbcCredential(jdbcUser, jdbcPassword);

spark-connector/spark-common/src/test/java/org/apache/gravitino/spark/connector/integration/test/jdbc/SparkJdbcPostgreSqlCatalogIT.java

@@ -22,6 +22,8 @@
 
 import com.google.common.collect.Maps;
 import java.util.Map;
+import org.apache.gravitino.credential.CredentialConstants;
+import org.apache.gravitino.credential.JdbcCredential;
 import org.apache.gravitino.integration.test.container.ContainerSuite;
 import org.apache.gravitino.spark.connector.integration.test.SparkCommonIT;
 import org.apache.gravitino.spark.connector.integration.test.util.SparkTableInfoChecker;
@@ -115,6 +117,8 @@ protected Map<String, String> getCatalogConfigs() {
     catalogProperties.put(JdbcPropertiesConstants.GRAVITINO_JDBC_DRIVER, this.pgDriver);
     catalogProperties.put(
         JdbcPropertiesConstants.GRAVITINO_JDBC_DATABASE, PG_CATALOG_PG_IT.toString());
+    catalogProperties.put(
+        CredentialConstants.CREDENTIAL_PROVIDERS, JdbcCredential.JDBC_CREDENTIAL_TYPE);
     return catalogProperties;
   }
 }

trino-connector/trino-connector/src/test/java/org/apache/gravitino/trino/connector/catalog/jdbc/postgresql/TestPostgreSQLConnectorAdapter.java

@@ -0,0 +1,67 @@
+/*
+ * 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.gravitino.trino.connector.catalog.jdbc.postgresql;
+
+import java.util.Collections;
+import java.util.Map;
+import org.apache.gravitino.credential.Credential;
+import org.apache.gravitino.credential.JdbcCredential;
+import org.apache.gravitino.trino.connector.catalog.jdbc.JDBCCatalogPropertyConverter;
+import org.apache.gravitino.trino.connector.metadata.GravitinoCatalog;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+
+/** Unit tests for {@link PostgreSQLConnectorAdapter#buildInternalConnectorConfig}. */
+public class TestPostgreSQLConnectorAdapter {
+
+  private static final String JDBC_URL = "jdbc:postgresql://localhost:5432/";
+
+  private GravitinoCatalog catalogWithProps(Map<String, String> props) {
+    return new GravitinoCatalog("metalake", "jdbc-postgresql", "test", props, 0L);
+  }
+
+  @Test
+  void testCredentialVendingOverridesProperties() throws Exception {
+    GravitinoCatalog catalog = catalogWithProps(Collections.singletonMap("jdbc-url", JDBC_URL));
+    Credential[] credentials = {new JdbcCredential("vended-user", "vended-password")};
+
+    Map<String, String> config =
+        new PostgreSQLConnectorAdapter().buildInternalConnectorConfig(catalog, credentials);
+
+    Assertions.assertEquals(
+        "vended-user", config.get(JDBCCatalogPropertyConverter.JDBC_CONNECTION_USER_KEY));
+    Assertions.assertEquals(
+        "vended-password", config.get(JDBCCatalogPropertyConverter.JDBC_CONNECTION_PASSWORD_KEY));
+  }
+
+  @Test
+  void testFallsBackToPropertiesWhenNoCredentials() throws Exception {
+    Map<String, String> props =
+        Map.of("jdbc-url", JDBC_URL, "jdbc-user", "prop-user", "jdbc-password", "prop-password");
+    GravitinoCatalog catalog = catalogWithProps(props);
+
+    Map<String, String> config =
+        new PostgreSQLConnectorAdapter().buildInternalConnectorConfig(catalog, new Credential[0]);
+
+    Assertions.assertEquals(
+        "prop-user", config.get(JDBCCatalogPropertyConverter.JDBC_CONNECTION_USER_KEY));
+    Assertions.assertEquals(
+        "prop-password", config.get(JDBCCatalogPropertyConverter.JDBC_CONNECTION_PASSWORD_KEY));
+  }
+}