Resolved more comments

Author: XJDKC

polaris-core/src/main/java/org/apache/polaris/core/entity/CatalogEntity.java

@@ -109,6 +109,10 @@ public static CatalogEntity fromCatalog(RealmConfig realmConfig, Catalog catalog
     return builder.build();
   }
 
+  public Catalog asCatalog() {
+    return this.asCatalog(null);
+  }
+
   public Catalog asCatalog(ServiceIdentityRegistry serviceIdentityRegistry) {
     Map<String, String> internalProperties = getInternalPropertiesAsMap();
     Catalog.TypeEnum catalogType =
@@ -120,6 +124,12 @@ public Catalog asCatalog(ServiceIdentityRegistry serviceIdentityRegistry) {
         CatalogProperties.builder(propertiesMap.get(DEFAULT_BASE_LOCATION_KEY))
             .putAll(propertiesMap)
             .build();
+
+    // Right now, only external catalog may use ServiceIdentityRegistry to resolve identity
+    Preconditions.checkState(
+        catalogType != Catalog.TypeEnum.EXTERNAL || serviceIdentityRegistry != null,
+        "catalog needs ServiceIdentityRegistry to resolve service identities",
+        Catalog.TypeEnum.EXTERNAL);
     return catalogType == Catalog.TypeEnum.EXTERNAL
         ? ExternalCatalog.builder()
             .setType(Catalog.TypeEnum.EXTERNAL)

polaris-core/src/main/java/org/apache/polaris/core/identity/resolved/ResolvedAwsIamServiceIdentity.java

@@ -18,8 +18,6 @@
  */
 package org.apache.polaris.core.identity.resolved;
 
-import com.google.common.base.Supplier;
-import com.google.common.base.Suppliers;
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import org.apache.polaris.core.admin.model.AwsIamServiceIdentityInfo;
@@ -31,7 +29,6 @@
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
 import software.amazon.awssdk.services.sts.StsClient;
-import software.amazon.awssdk.services.sts.StsClientBuilder;
 
 /**
  * Represents a fully resolved AWS IAM service identity, including the associated IAM ARN and
@@ -93,14 +90,4 @@ public ResolvedAwsIamServiceIdentity(
         .setIamArn(getIamArn())
         .build();
   }
-
-  /** Returns a memoized supplier for creating an STS client using the resolved credentials. */
-  public @Nonnull Supplier<StsClient> stsClientSupplier() {
-    return Suppliers.memoize(
-        () -> {
-          StsClientBuilder stsClientBuilder =
-              StsClient.builder().credentialsProvider(getAwsCredentialsProvider());
-          return stsClientBuilder.build();
-        });
-  }
 }

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

@@ -23,7 +23,7 @@
 import java.util.Optional;
 import org.apache.polaris.core.identity.ServiceIdentityType;
 import org.apache.polaris.core.identity.resolved.ResolvedAwsIamServiceIdentity;
-import org.apache.polaris.service.identity.registry.DefaultServiceIdentityRegistry;
+import org.apache.polaris.core.secrets.ServiceSecretReference;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.AwsSessionCredentials;
@@ -60,23 +60,32 @@ public interface AwsIamServiceIdentityConfiguration extends ResolvableServiceIde
    */
   Optional<String> sessionToken();
 
+  /**
+   * Returns the type of service identity represented by this configuration, which is always {@link
+   * ServiceIdentityType#AWS_IAM}.
+   *
+   * @return the AWS IAM service identity type
+   */
+  @Override
+  default ServiceIdentityType getType() {
+    return ServiceIdentityType.AWS_IAM;
+  }
+
   /**
    * Resolves this configuration into a {@link ResolvedAwsIamServiceIdentity} if the IAM ARN is
    * present.
    *
    * @return the resolved identity, or an empty optional if the ARN is missing
    */
   @Override
-  default Optional<ResolvedAwsIamServiceIdentity> resolve(@Nonnull String realmIdentifier) {
+  default Optional<ResolvedAwsIamServiceIdentity> resolve(
+      @Nonnull ServiceSecretReference serviceIdentityReference) {
     if (iamArn() == null) {
       return Optional.empty();
     } else {
       return Optional.of(
           new ResolvedAwsIamServiceIdentity(
-              DefaultServiceIdentityRegistry.buildIdentityInfoReference(
-                  realmIdentifier, ServiceIdentityType.AWS_IAM),
-              iamArn(),
-              awsCredentialsProvider()));
+              serviceIdentityReference, iamArn(), awsCredentialsProvider()));
     }
   }
 

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

@@ -21,7 +21,9 @@
 
 import jakarta.annotation.Nonnull;
 import java.util.Optional;
+import org.apache.polaris.core.identity.ServiceIdentityType;
 import org.apache.polaris.core.identity.resolved.ResolvedServiceIdentity;
+import org.apache.polaris.core.secrets.ServiceSecretReference;
 
 /**
  * Represents a service identity configuration that can be resolved into a fully initialized {@link
@@ -32,13 +34,23 @@
  * Polaris-managed service identity.
  */
 public interface ResolvableServiceIdentityConfiguration {
+  /**
+   * Returns the type of service identity represented by this configuration.
+   *
+   * @return the service identity type, or {@link ServiceIdentityType#NULL_TYPE} if not specified
+   */
+  default ServiceIdentityType getType() {
+    return ServiceIdentityType.NULL_TYPE;
+  }
+
   /**
    * Attempts to resolve this configuration into a {@link ResolvedServiceIdentity}.
    *
    * @return an optional resolved service identity, or empty if resolution fails or is not
    *     configured
    */
-  default Optional<? extends ResolvedServiceIdentity> resolve(@Nonnull String realmIdentifier) {
+  default Optional<? extends ResolvedServiceIdentity> resolve(
+      @Nonnull ServiceSecretReference serviceIdentityReference) {
     return Optional.empty();
   }
 }

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

@@ -28,6 +28,7 @@
 import java.util.Optional;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.identity.resolved.ResolvedServiceIdentity;
+import org.apache.polaris.service.identity.registry.DefaultServiceIdentityRegistry;
 
 /**
  * Represents the service identity configuration for one or more realms.
@@ -84,7 +85,9 @@ default List<? extends ResolvedServiceIdentity> resolveServiceIdentities(
     return entry.config().serviceIdentityConfigurations().stream()
         .map(
             resolvableServiceIdentityConfiguration ->
-                resolvableServiceIdentityConfiguration.resolve(entry.realm()))
+                resolvableServiceIdentityConfiguration.resolve(
+                    DefaultServiceIdentityRegistry.buildIdentityInfoReference(
+                        entry.realm(), resolvableServiceIdentityConfiguration.getType())))
         .flatMap(Optional::stream)
         .toList();
   }