[Catalog Federation] Add Connection Credential Vendors for Other Auth Types (#2782)

Add Connection Credential Vendors for Other Auth Types

This change is a prerequisite for enabling connection credential caching.
By making PolarisCredentialManager the central entry point for obtaining connection credentials, we can introduce caching cleanly and manage all credential flows in a consistent way.

Author: XJDKC

extensions/federation/hadoop/src/main/java/org/apache/polaris/extensions/federation/hadoop/HadoopFederatedCatalogFactory.java

@@ -31,7 +31,6 @@
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.connection.hadoop.HadoopConnectionConfigInfoDpo;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -44,7 +43,6 @@ public class HadoopFederatedCatalogFactory implements ExternalCatalogFactory {
   @Override
   public Catalog createCatalog(
       ConnectionConfigInfoDpo connectionConfigInfoDpo,
-      UserSecretsManager userSecretsManager,
       PolarisCredentialManager polarisCredentialManager) {
     // Currently, Polaris supports Hadoop federation only via IMPLICIT authentication.
     // Hence, prior to initializing the configuration, ensure that the catalog uses
@@ -59,15 +57,13 @@ public Catalog createCatalog(
     String warehouse = ((HadoopConnectionConfigInfoDpo) connectionConfigInfoDpo).getWarehouse();
     HadoopCatalog hadoopCatalog = new HadoopCatalog(conf, warehouse);
     hadoopCatalog.initialize(
-        warehouse,
-        connectionConfigInfoDpo.asIcebergCatalogProperties(
-            userSecretsManager, polarisCredentialManager));
+        warehouse, connectionConfigInfoDpo.asIcebergCatalogProperties(polarisCredentialManager));
     return hadoopCatalog;
   }
 
   @Override
   public GenericTableCatalog createGenericCatalog(
-      ConnectionConfigInfoDpo connectionConfig, UserSecretsManager userSecretsManager) {
+      ConnectionConfigInfoDpo connectionConfig, PolarisCredentialManager polarisCredentialManager) {
     // TODO implement
     throw new UnsupportedOperationException(
         "Generic table federation to this catalog is not supported.");

extensions/federation/hive/src/main/java/org/apache/polaris/extensions/federation/hive/HiveFederatedCatalogFactory.java

@@ -30,7 +30,6 @@
 import org.apache.polaris.core.connection.ConnectionType;
 import org.apache.polaris.core.connection.hive.HiveConnectionConfigInfoDpo;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -43,7 +42,6 @@ public class HiveFederatedCatalogFactory implements ExternalCatalogFactory {
   @Override
   public Catalog createCatalog(
       ConnectionConfigInfoDpo connectionConfigInfoDpo,
-      UserSecretsManager userSecretsManager,
       PolarisCredentialManager polarisCredentialManager) {
     // Currently, Polaris supports Hive federation only via IMPLICIT authentication.
     // Hence, prior to initializing the configuration, ensure that the catalog uses
@@ -72,15 +70,13 @@ public Catalog createCatalog(
     // Kerberos instances are not suitable because Kerberos ties a single identity to the server.
     HiveCatalog hiveCatalog = new HiveCatalog();
     hiveCatalog.initialize(
-        warehouse,
-        connectionConfigInfoDpo.asIcebergCatalogProperties(
-            userSecretsManager, polarisCredentialManager));
+        warehouse, connectionConfigInfoDpo.asIcebergCatalogProperties(polarisCredentialManager));
     return hiveCatalog;
   }
 
   @Override
   public GenericTableCatalog createGenericCatalog(
-      ConnectionConfigInfoDpo connectionConfig, UserSecretsManager userSecretsManager) {
+      ConnectionConfigInfoDpo connectionConfig, PolarisCredentialManager polarisCredentialManager) {
     // TODO implement
     throw new UnsupportedOperationException(
         "Generic table federation to this catalog is not supported.");

polaris-core/src/main/java/org/apache/polaris/core/catalog/ExternalCatalogFactory.java

@@ -21,7 +21,6 @@
 import org.apache.iceberg.catalog.Catalog;
 import org.apache.polaris.core.connection.ConnectionConfigInfoDpo;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * Factory interface for creating external catalog handles based on connection configuration.
@@ -35,25 +34,23 @@ public interface ExternalCatalogFactory {
    * Creates a catalog handle for the given connection configuration.
    *
    * @param connectionConfig the connection configuration
-   * @param userSecretsManager the user secrets manager for handling user-provided credentials
-   * @param polarisCredentialManager the credential manager for generating temporary credentials
+   * @param polarisCredentialManager the credential manager for generating connection credentials
    *     that Polaris uses to access external systems
    * @return the initialized catalog
    * @throws IllegalStateException if the connection configuration is invalid
    */
   Catalog createCatalog(
-      ConnectionConfigInfoDpo connectionConfig,
-      UserSecretsManager userSecretsManager,
-      PolarisCredentialManager polarisCredentialManager);
+      ConnectionConfigInfoDpo connectionConfig, PolarisCredentialManager polarisCredentialManager);
 
   /**
    * Creates a generic table catalog for the given connection configuration.
    *
    * @param connectionConfig the connection configuration
-   * @param userSecretsManager the user secrets manager for handling credentials
+   * @param polarisCredentialManager the credential manager for generating connection credentials
+   *     that Polaris uses to access external systems
    * @return the initialized catalog
    * @throws IllegalStateException if the connection configuration is invalid
    */
   GenericTableCatalog createGenericCatalog(
-      ConnectionConfigInfoDpo connectionConfig, UserSecretsManager userSecretsManager);
+      ConnectionConfigInfoDpo connectionConfig, PolarisCredentialManager polarisCredentialManager);
 }

polaris-core/src/main/java/org/apache/polaris/core/connection/BearerAuthenticationParametersDpo.java

@@ -21,13 +21,9 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.google.common.base.MoreObjects;
 import jakarta.annotation.Nonnull;
-import java.util.Map;
-import org.apache.iceberg.rest.auth.OAuth2Properties;
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.BearerAuthenticationParameters;
-import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.SecretReference;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to BearerAuthenticationParameters defined in the API
@@ -49,13 +45,6 @@ public BearerAuthenticationParametersDpo(
     return bearerTokenReference;
   }
 
-  @Override
-  public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
-    String bearerToken = secretsManager.readSecret(getBearerTokenReference());
-    return Map.of(OAuth2Properties.TOKEN, bearerToken);
-  }
-
   @Override
   public @Nonnull AuthenticationParameters asAuthenticationParametersModel() {
     return BearerAuthenticationParameters.builder()

polaris-core/src/main/java/org/apache/polaris/core/connection/ImplicitAuthenticationParametersDpo.java

@@ -24,7 +24,6 @@
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.ImplicitAuthenticationParameters;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to ImplicitAuthenticationParameters defined in the
@@ -38,7 +37,9 @@ public ImplicitAuthenticationParametersDpo() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
+      PolarisCredentialManager credentialManager) {
+    // Return only metadata properties - credentials are handled by ConnectionCredentialVendor
+    // Implicit auth has no metadata properties
     return Map.of();
   }
 

polaris-core/src/main/java/org/apache/polaris/core/connection/OAuthClientCredentialsParametersDpo.java

@@ -37,7 +37,6 @@
 import org.apache.polaris.core.admin.model.OAuthClientCredentialsParameters;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
 import org.apache.polaris.core.secrets.SecretReference;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to OAuthClientCredentialsParameters defined in the
@@ -97,20 +96,14 @@ public OAuthClientCredentialsParametersDpo(
         Objects.requireNonNullElse(scopes, List.of(OAuth2Properties.CATALOG_SCOPE)));
   }
 
-  @JsonIgnore
-  private @Nonnull String getCredentialAsConcatenatedString(UserSecretsManager secretsManager) {
-    String clientSecret = secretsManager.readSecret(getClientSecretReference());
-    return COLON_JOINER.join(clientId, clientSecret);
-  }
-
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
+      PolarisCredentialManager credentialManager) {
+    // Return only metadata properties - credentials are handled by ConnectionCredentialVendor
     HashMap<String, String> properties = new HashMap<>();
     if (getTokenUri() != null) {
       properties.put(OAuth2Properties.OAUTH2_SERVER_URI, getTokenUri());
     }
-    properties.put(OAuth2Properties.CREDENTIAL, getCredentialAsConcatenatedString(secretsManager));
     properties.put(OAuth2Properties.SCOPE, getScopesAsString());
     return properties;
   }

polaris-core/src/main/java/org/apache/polaris/core/connection/SigV4AuthenticationParametersDpo.java

@@ -29,7 +29,6 @@
 import org.apache.polaris.core.admin.model.AuthenticationParameters;
 import org.apache.polaris.core.admin.model.SigV4AuthenticationParameters;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to SigV4AuthenticationParameters defined in the API
@@ -95,14 +94,14 @@ public SigV4AuthenticationParametersDpo(
   @Nonnull
   @Override
   public Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
+      PolarisCredentialManager credentialManager) {
+    // Return only metadata properties - credentials are handled by ConnectionCredentialVendor
     ImmutableMap.Builder<String, String> builder = ImmutableMap.builder();
     builder.put(AuthProperties.AUTH_TYPE, AuthProperties.AUTH_TYPE_SIGV4);
     builder.put(AwsProperties.REST_SIGNER_REGION, getSigningRegion());
     if (getSigningName() != null) {
       builder.put(AwsProperties.REST_SIGNING_NAME, getSigningName());
     }
-    // Connection credentials are handled by ConnectionConfigInfoDpo
     return builder.build();
   }
 

polaris-core/src/main/java/org/apache/polaris/core/connection/hadoop/HadoopConnectionConfigInfoDpo.java

@@ -35,7 +35,6 @@
 import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to {@link
@@ -73,16 +72,14 @@ public String toString() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager) {
+      PolarisCredentialManager credentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     properties.put(CatalogProperties.URI, getUri());
     if (getWarehouse() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getWarehouse());
     }
-    // Add authentication-specific properties
-    properties.putAll(
-        getAuthenticationParameters()
-            .asIcebergCatalogProperties(secretsManager, credentialManager));
+    // Add authentication-specific metadata (non-credential properties)
+    properties.putAll(getAuthenticationParameters().asIcebergCatalogProperties(credentialManager));
     // Add connection credentials from Polaris credential manager
     ConnectionCredentials connectionCredentials = credentialManager.getConnectionCredentials(this);
     properties.putAll(connectionCredentials.credentials());

polaris-core/src/main/java/org/apache/polaris/core/connection/hive/HiveConnectionConfigInfoDpo.java

@@ -35,7 +35,6 @@
 import org.apache.polaris.core.credentials.connection.ConnectionCredentials;
 import org.apache.polaris.core.identity.dpo.ServiceIdentityInfoDpo;
 import org.apache.polaris.core.identity.provider.ServiceIdentityProvider;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * The internal persistence-object counterpart to {@link
@@ -72,17 +71,16 @@ public String toString() {
 
   @Override
   public @Nonnull Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager polarisCredentialManager) {
+      PolarisCredentialManager polarisCredentialManager) {
     HashMap<String, String> properties = new HashMap<>();
     properties.put(CatalogProperties.URI, getUri());
     if (getWarehouse() != null) {
       properties.put(CatalogProperties.WAREHOUSE_LOCATION, getWarehouse());
     }
     if (getAuthenticationParameters() != null) {
-      // Add authentication-specific properties
+      // Add authentication-specific metadata (non-credential properties)
       properties.putAll(
-          getAuthenticationParameters()
-              .asIcebergCatalogProperties(secretsManager, polarisCredentialManager));
+          getAuthenticationParameters().asIcebergCatalogProperties(polarisCredentialManager));
       // Add connection credentials from Polaris credential manager
       ConnectionCredentials connectionCredentials =
           polarisCredentialManager.getConnectionCredentials(this);

polaris-core/src/main/java/org/apache/polaris/core/connection/iceberg/IcebergCatalogPropertiesProvider.java

@@ -21,7 +21,6 @@
 import jakarta.annotation.Nonnull;
 import java.util.Map;
 import org.apache.polaris.core.credentials.PolarisCredentialManager;
-import org.apache.polaris.core.secrets.UserSecretsManager;
 
 /**
  * Configuration wrappers which ultimately translate their contents into Iceberg properties and
@@ -31,6 +30,8 @@
  */
 public interface IcebergCatalogPropertiesProvider {
   @Nonnull
-  Map<String, String> asIcebergCatalogProperties(
-      UserSecretsManager secretsManager, PolarisCredentialManager credentialManager);
+  default Map<String, String> asIcebergCatalogProperties(
+      PolarisCredentialManager credentialManager) {
+    return Map.of();
+  }
 }