Move some logic to ServiceIdentityConfiguration

Author: XJDKC

polaris-core/src/main/java/org/apache/polaris/core/identity/registry/ServiceIdentityRegistry.java

@@ -44,7 +44,7 @@ public interface ServiceIdentityRegistry {
    * @param serviceIdentityType The type of service identity (e.g., AWS_IAM).
    * @return A new {@link ServiceIdentityInfoDpo} representing the discovered service identity.
    */
-  ServiceIdentityInfoDpo discoverServiceIdentity(ServiceIdentityType serviceIdentityType);
+  Optional<ServiceIdentityInfoDpo> discoverServiceIdentity(ServiceIdentityType serviceIdentityType);
 
   /**
    * Resolves the given service identity by retrieving the actual credential or secret referenced by

runtime/defaults/src/main/resources/application.properties

@@ -113,7 +113,7 @@ polaris.realm-context.require-header=false
 
 polaris.features."ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING"=false
 polaris.features."SUPPORTED_CATALOG_STORAGE_TYPES"=["S3","GCS","AZURE"]
-polaris.features."ENABLE_CATALOG_FEDERATION"=true
+# polaris.features."ENABLE_CATALOG_FEDERATION"=true
 polaris.features."SUPPORTED_CATALOG_CONNECTION_TYPES"=["ICEBERG_REST"]
 
 # realm overrides
@@ -200,7 +200,7 @@ polaris.oidc.principal-roles-mapper.type=default
 
 # Polaris Service Identity Config
 # Default identity (can be overridden in per realm)
-polaris.service-identity.aws-iam.iam-arn=arn:aws:iam::123456789012:user/polaris-iam-user
+# polaris.service-identity.aws-iam.iam-arn=arn:aws:iam::123456789012:user/polaris-iam-user
 # polaris.service-identity.aws-iam.access-key-id=accessKeyId
 # polaris.service-identity.aws-iam.secret-access-key=secretAccessKey
 # polaris.service-identity.aws-iam.session-token=sessionToken

runtime/service/src/main/java/org/apache/polaris/service/admin/PolarisAdminService.java

@@ -784,17 +784,31 @@ public PolarisEntity createCatalog(CreateCatalogRequest catalogRequest) {
               "Implicit authentication based catalog federation is not supported.");
         }
 
-        ServiceIdentityInfoDpo serviceIdentityInfo = null;
+        // Discover service identity if needed for the authentication type.
+        Optional<ServiceIdentityInfoDpo> serviceIdentityInfoDpoOptional = Optional.empty();
         if (connectionConfigInfo.getAuthenticationParameters().getAuthenticationType()
             == AuthenticationParameters.AuthenticationTypeEnum.SIGV4) {
-          serviceIdentityInfo =
+          serviceIdentityInfoDpoOptional =
               serviceIdentityRegistry.discoverServiceIdentity(ServiceIdentityType.AWS_IAM);
+          if (serviceIdentityInfoDpoOptional.isEmpty()) {
+            throw new IllegalStateException(
+                String.format(
+                    "Cannot create Catalog %s. Failed to discover %s service identity for %s authentication",
+                    entity.getName(),
+                    ServiceIdentityType.AWS_IAM.name(),
+                    connectionConfigInfo
+                        .getAuthenticationParameters()
+                        .getAuthenticationType()
+                        .name()));
+          }
         }
 
         entity =
             new CatalogEntity.Builder(entity)
                 .setConnectionConfigInfoDpoWithSecrets(
-                    connectionConfigInfo, processedSecretReferences, serviceIdentityInfo)
+                    connectionConfigInfo,
+                    processedSecretReferences,
+                    serviceIdentityInfoDpoOptional.orElse(null))
                 .build();
       }
     }

runtime/service/src/main/java/org/apache/polaris/service/identity/AwsIamServiceIdentityConfiguration.java

@@ -21,8 +21,14 @@
 
 import jakarta.annotation.Nonnull;
 import java.util.Optional;
+import org.apache.polaris.core.identity.ServiceIdentityType;
 import org.apache.polaris.core.identity.resolved.ResolvedAwsIamServiceIdentity;
-import software.amazon.awssdk.auth.credentials.*;
+import org.apache.polaris.service.identity.registry.DefaultServiceIdentityRegistry;
+import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 
 /**
  * Configuration for an AWS IAM service identity used by Polaris to access AWS services.
@@ -61,11 +67,16 @@ public interface AwsIamServiceIdentityConfiguration extends ResolvableServiceIde
    * @return the resolved identity, or an empty optional if the ARN is missing
    */
   @Override
-  default Optional<ResolvedAwsIamServiceIdentity> resolve() {
+  default Optional<ResolvedAwsIamServiceIdentity> resolve(@Nonnull String realmIdentifier) {
     if (iamArn() == null) {
       return Optional.empty();
     } else {
-      return Optional.of(new ResolvedAwsIamServiceIdentity(iamArn(), awsCredentialsProvider()));
+      return Optional.of(
+          new ResolvedAwsIamServiceIdentity(
+              DefaultServiceIdentityRegistry.buildIdentityInfoReference(
+                  realmIdentifier, ServiceIdentityType.AWS_IAM),
+              iamArn(),
+              awsCredentialsProvider()));
     }
   }
 

runtime/service/src/main/java/org/apache/polaris/service/identity/ResolvableServiceIdentityConfiguration.java

@@ -19,6 +19,7 @@
 
 package org.apache.polaris.service.identity;
 
+import jakarta.annotation.Nonnull;
 import java.util.Optional;
 import org.apache.polaris.core.identity.resolved.ResolvedServiceIdentity;
 
@@ -37,5 +38,7 @@ public interface ResolvableServiceIdentityConfiguration {
    * @return an optional resolved service identity, or empty if resolution fails or is not
    *     configured
    */
-  Optional<? extends ResolvedServiceIdentity> resolve();
+  default Optional<? extends ResolvedServiceIdentity> resolve(@Nonnull String realmIdentifier) {
+    return Optional.empty();
+  }
 }

runtime/service/src/main/java/org/apache/polaris/service/identity/ServiceIdentityConfiguration.java

@@ -23,8 +23,11 @@
 import io.smallrye.config.WithDefaults;
 import io.smallrye.config.WithParentName;
 import io.smallrye.config.WithUnnamedKey;
+import java.util.List;
 import java.util.Map;
+import java.util.Optional;
 import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.identity.resolved.ResolvedServiceIdentity;
 
 /**
  * Represents the service identity configuration for one or more realms.
@@ -54,48 +57,38 @@ public interface ServiceIdentityConfiguration {
   Map<String, RealmServiceIdentityConfiguration> realms();
 
   /**
-   * Returns the service identity configuration for the given {@link RealmContext}. Falls back to
-   * the default if the realm is not explicitly configured.
-   *
-   * @param realmContext the realm context
-   * @return the matching or default realm configuration
+   * Resolves the actual realm configuration entry (identifier + config) to use for the given
+   * context. Falls back to the default if the specified realm is not configured.
    */
-  default RealmServiceIdentityConfiguration forRealm(RealmContext realmContext) {
+  default RealmConfigEntry forRealm(RealmContext realmContext) {
     return forRealm(realmContext.getRealmIdentifier());
   }
 
   /**
-   * Returns the service identity configuration for the given realm identifier. Falls back to the
-   * default if the realm is not explicitly configured.
-   *
-   * @param realmIdentifier the identifier of the realm
-   * @return the matching or default realm configuration
+   * Resolves the actual realm configuration entry (identifier + config) for the given realm
+   * identifier. Falls back to the default if the specified realm is not configured.
    */
-  default RealmServiceIdentityConfiguration forRealm(String realmIdentifier) {
-    return realms().containsKey(realmIdentifier)
-        ? realms().get(realmIdentifier)
-        : realms().get(DEFAULT_REALM_KEY);
+  default RealmConfigEntry forRealm(String realmIdentifier) {
+    String resolvedRealmIdentifier =
+        realms().containsKey(realmIdentifier) ? realmIdentifier : DEFAULT_REALM_KEY;
+    return new RealmConfigEntry(resolvedRealmIdentifier, realms().get(resolvedRealmIdentifier));
   }
 
   /**
-   * Returns the actual key of the service identity configuration to use for the given {@link
-   * RealmContext}, falling back to the default if the specified realm is not configured.
-   *
-   * @param realmContext the realm context
-   * @return the actual realm identifier to use
+   * Resolves and returns the list of {@link ResolvedServiceIdentity} objects for the given realm.
    */
-  default String resolveRealm(RealmContext realmContext) {
-    return resolveRealm(realmContext.getRealmIdentifier());
-  }
+  default List<? extends ResolvedServiceIdentity> resolveServiceIdentities(
+      RealmContext realmContext) {
+    RealmConfigEntry entry = forRealm(realmContext);
 
-  /**
-   * Returns the actual key of the service identity configuration to use for the given realm
-   * identifier, falling back to the default if the specified realm is not configured.
-   *
-   * @param realmIdentifier the identifier of the realm
-   * @return the actual realm identifier to use
-   */
-  default String resolveRealm(String realmIdentifier) {
-    return realms().containsKey(realmIdentifier) ? realmIdentifier : DEFAULT_REALM_KEY;
+    return entry.config().serviceIdentityConfigurations().stream()
+        .map(
+            resolvableServiceIdentityConfiguration ->
+                resolvableServiceIdentityConfiguration.resolve(entry.realm()))
+        .flatMap(Optional::stream)
+        .toList();
   }
+
+  /** A pairing of a resolved realm identifier and its associated configuration. */
+  record RealmConfigEntry(String realm, RealmServiceIdentityConfiguration config) {}
 }

runtime/service/src/main/java/org/apache/polaris/service/identity/registry/DefaultServiceIdentityRegistry.java

@@ -31,8 +31,6 @@
 import org.apache.polaris.core.identity.registry.ServiceIdentityRegistry;
 import org.apache.polaris.core.identity.resolved.ResolvedServiceIdentity;
 import org.apache.polaris.core.secrets.ServiceSecretReference;
-import org.apache.polaris.service.identity.RealmServiceIdentityConfiguration;
-import org.apache.polaris.service.identity.ResolvableServiceIdentityConfiguration;
 import org.apache.polaris.service.identity.ServiceIdentityConfiguration;
 
 /**
@@ -80,20 +78,8 @@ public DefaultServiceIdentityRegistry(
   @Inject
   public DefaultServiceIdentityRegistry(
       RealmContext realmContext, ServiceIdentityConfiguration serviceIdentityConfiguration) {
-    String serviceIdentityConfigKey = serviceIdentityConfiguration.resolveRealm(realmContext);
-    RealmServiceIdentityConfiguration realmServiceIdentityConfiguration =
-        serviceIdentityConfiguration.forRealm(realmContext);
-
     this.resolvedServiceIdentities =
-        realmServiceIdentityConfiguration.serviceIdentityConfigurations().stream()
-            .map(ResolvableServiceIdentityConfiguration::resolve)
-            .flatMap(Optional::stream)
-            .peek(
-                // Set the identity info reference for each resolved identity
-                identity ->
-                    identity.setIdentityInfoReference(
-                        buildIdentityInfoReference(
-                            serviceIdentityConfigKey, identity.getIdentityType())))
+        serviceIdentityConfiguration.resolveServiceIdentities(realmContext).stream()
             .collect(
                 // Collect to an EnumMap, grouping by ServiceIdentityType
                 Collectors.toMap(
@@ -111,14 +97,14 @@ public DefaultServiceIdentityRegistry(
   }
 
   @Override
-  public ServiceIdentityInfoDpo discoverServiceIdentity(ServiceIdentityType serviceIdentityType) {
+  public Optional<ServiceIdentityInfoDpo> discoverServiceIdentity(
+      ServiceIdentityType serviceIdentityType) {
     ResolvedServiceIdentity resolvedServiceIdentity =
         resolvedServiceIdentities.get(serviceIdentityType);
     if (resolvedServiceIdentity == null) {
-      throw new IllegalArgumentException(
-          "Service identity type not supported: " + serviceIdentityType);
+      return Optional.empty();
     }
-    return resolvedServiceIdentity.asServiceIdentityInfoDpo();
+    return Optional.of(resolvedServiceIdentity.asServiceIdentityInfoDpo());
   }
 
   @Override
@@ -147,7 +133,7 @@ public EnumMap<ServiceIdentityType, ResolvedServiceIdentity> getResolvedServiceI
    * @param type the service identity type
    * @return the constructed service secret reference
    */
-  private ServiceSecretReference buildIdentityInfoReference(
+  public static ServiceSecretReference buildIdentityInfoReference(
       String realm, ServiceIdentityType type) {
     // urn:polaris-service-secret:default-identity-registry:<realm>:<type>
     return new ServiceSecretReference(

runtime/service/src/test/java/org/apache/polaris/service/identity/registry/DefaultServiceIdentityRegistryTest.java

@@ -37,6 +37,9 @@
 import org.assertj.core.api.Assertions;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
+import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 
 @QuarkusTest
 @TestProfile(DefaultServiceIdentityRegistryTest.Profile.class)
@@ -77,8 +80,10 @@ void testServiceIdentityConfiguration() {
         .isEqualTo(2);
 
     // Check the default realm configuration
-    RealmServiceIdentityConfiguration defaultConfig =
+    ServiceIdentityConfiguration.RealmConfigEntry defaultConfigEntry =
         serviceIdentityConfiguration.forRealm(DEFAULT_REALM_KEY);
+    Assertions.assertThat(defaultConfigEntry.realm()).isEqualTo(DEFAULT_REALM_KEY);
+    RealmServiceIdentityConfiguration defaultConfig = defaultConfigEntry.config();
     Assertions.assertThat(defaultConfig.awsIamServiceIdentity().isPresent()).isTrue();
     Assertions.assertThat(defaultConfig.awsIamServiceIdentity().get().iamArn())
         .isEqualTo("arn:aws:iam::123456789012:user/polaris-default-iam-user");
@@ -87,8 +92,10 @@ void testServiceIdentityConfiguration() {
     Assertions.assertThat(defaultConfig.awsIamServiceIdentity().get().sessionToken()).isEmpty();
 
     // Check the my-realm configuration
-    RealmServiceIdentityConfiguration myRealmConfig =
+    ServiceIdentityConfiguration.RealmConfigEntry myRealmConfigEntry =
         serviceIdentityConfiguration.forRealm(MY_REALM_KEY);
+    Assertions.assertThat(myRealmConfigEntry.realm()).isEqualTo(MY_REALM_KEY);
+    RealmServiceIdentityConfiguration myRealmConfig = myRealmConfigEntry.config();
     Assertions.assertThat(myRealmConfig.awsIamServiceIdentity().isPresent()).isTrue();
     Assertions.assertThat(myRealmConfig.awsIamServiceIdentity().get().iamArn())
         .isEqualTo("arn:aws:iam::123456789012:user/polaris-iam-user");
@@ -100,8 +107,10 @@ void testServiceIdentityConfiguration() {
         .isEqualTo(Optional.of("session-token"));
 
     // Check the unexisting realm configuration
-    RealmServiceIdentityConfiguration otherConfig =
+    ServiceIdentityConfiguration.RealmConfigEntry otherConfigEntry =
         serviceIdentityConfiguration.forRealm("other-realm");
+    Assertions.assertThat(otherConfigEntry.realm()).isEqualTo(DEFAULT_REALM_KEY);
+    RealmServiceIdentityConfiguration otherConfig = otherConfigEntry.config();
     Assertions.assertThat(otherConfig.awsIamServiceIdentity().isPresent()).isTrue();
     Assertions.assertThat(otherConfig.awsIamServiceIdentity().get().iamArn())
         .isEqualTo("arn:aws:iam::123456789012:user/polaris-default-iam-user");
@@ -131,9 +140,10 @@ void testRealmServiceIdentityConfigToResolvedServiceIdentity() {
         .isEqualTo(
             new ServiceSecretReference(
                 "urn:polaris-secret:default-identity-registry:system:default:AWS_IAM", Map.of()));
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getAccessKeyId()).isNull();
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSecretAccessKey()).isNull();
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSessionToken()).isNull();
+    Assertions.assertThat(
+            resolvedAwsIamServiceIdentity.getAwsCredentialsProvider()
+                instanceof DefaultCredentialsProvider)
+        .isTrue();
 
     // Check the my-realm
     Mockito.when(realmContext.getRealmIdentifier()).thenReturn(MY_REALM_KEY);
@@ -153,12 +163,20 @@ void testRealmServiceIdentityConfigToResolvedServiceIdentity() {
         .isEqualTo(
             new ServiceSecretReference(
                 "urn:polaris-secret:default-identity-registry:my-realm:AWS_IAM", Map.of()));
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getAccessKeyId())
-        .isEqualTo("access-key-id");
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSecretAccessKey())
-        .isEqualTo("secret-access-key");
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSessionToken())
-        .isEqualTo("session-token");
+    Assertions.assertThat(
+            resolvedAwsIamServiceIdentity.getAwsCredentialsProvider()
+                instanceof StaticCredentialsProvider)
+        .isTrue();
+    StaticCredentialsProvider staticCredentialsProvider =
+        (StaticCredentialsProvider) resolvedAwsIamServiceIdentity.getAwsCredentialsProvider();
+    Assertions.assertThat(
+            staticCredentialsProvider.resolveCredentials() instanceof AwsSessionCredentials)
+        .isTrue();
+    AwsSessionCredentials awsSessionCredentials =
+        (AwsSessionCredentials) staticCredentialsProvider.resolveCredentials();
+    Assertions.assertThat(awsSessionCredentials.accessKeyId()).isEqualTo("access-key-id");
+    Assertions.assertThat(awsSessionCredentials.secretAccessKey()).isEqualTo("secret-access-key");
+    Assertions.assertThat(awsSessionCredentials.sessionToken()).isEqualTo("session-token");
 
     // Check the other realm which does not exist in the configuration, should fallback to default
     Mockito.when(realmContext.getRealmIdentifier()).thenReturn("other-realm");
@@ -177,8 +195,9 @@ void testRealmServiceIdentityConfigToResolvedServiceIdentity() {
         .isEqualTo(
             new ServiceSecretReference(
                 "urn:polaris-secret:default-identity-registry:system:default:AWS_IAM", Map.of()));
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getAccessKeyId()).isNull();
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSecretAccessKey()).isNull();
-    Assertions.assertThat(resolvedAwsIamServiceIdentity.getSessionToken()).isNull();
+    Assertions.assertThat(
+            resolvedAwsIamServiceIdentity.getAwsCredentialsProvider()
+                instanceof DefaultCredentialsProvider)
+        .isTrue();
   }
 }