Remove CallContext

since `PolarisCredentialVendor` (a sub-interface of `PolarisMetaStoreManager`) no longer has
a `CallContext` parameter we can see a domino effect:
- `StorageCredentialCache` no longer needs `CallContext`
- `TokenBroker` no longer needs `CallContext`
- `FileIOFactory` no longer needs `CallContext`
- `TaskFileIOSupplier` no longer needs `CallContext`
- `TaskHandler` and `TaskExecutor` no longer needs `CallContext`
- `ResolverFactory` and `ResolutionManifest` no longer needs `CallContext`

and eventually `PolarisCallContext` and `CallContext` are really unused and can be removed
or replaced by passing `RealmContext` and `RealmConfig` individually.

Author: XN137

polaris-core/src/main/java/org/apache/polaris/core/PolarisCallContext.java

@@ -30,11 +30,13 @@
 import java.util.Set;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.auth.PolarisPrincipal;
-import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
 import org.apache.polaris.core.entity.PolarisEntitySubType;
 import org.apache.polaris.core.entity.PolarisEntityType;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 import org.apache.polaris.core.persistence.ResolvedPolarisEntity;
 import org.slf4j.Logger;
@@ -52,7 +54,9 @@ public class PolarisResolutionManifest implements PolarisResolutionManifestCatal
   private static final Logger LOGGER = LoggerFactory.getLogger(PolarisResolutionManifest.class);
 
   private final ResolverFactory resolverFactory;
-  private final CallContext callContext;
+  private final RealmContext realmContext;
+  private final RealmConfig realmConfig;
+  private final PolarisMetaStoreManager metaStoreManager;
   private final SecurityContext securityContext;
   private final String catalogName;
   private final Resolver primaryResolver;
@@ -71,15 +75,20 @@ public class PolarisResolutionManifest implements PolarisResolutionManifestCatal
 
   public PolarisResolutionManifest(
       PolarisDiagnostics diagnostics,
-      CallContext callContext,
+      RealmContext realmContext,
+      RealmConfig realmConfig,
+      PolarisMetaStoreManager metaStoreManager,
       ResolverFactory resolverFactory,
       SecurityContext securityContext,
       String catalogName) {
-    this.callContext = callContext;
+    this.realmContext = realmContext;
+    this.realmConfig = realmConfig;
+    this.metaStoreManager = metaStoreManager;
     this.resolverFactory = resolverFactory;
     this.catalogName = catalogName;
     this.primaryResolver =
-        resolverFactory.createResolver(callContext, securityContext, catalogName);
+        resolverFactory.createResolver(
+            realmContext, realmConfig, metaStoreManager, securityContext, catalogName);
     this.diagnostics = diagnostics;
     this.diagnostics.checkNotNull(securityContext, "null_security_context_for_resolution_manifest");
     this.securityContext = securityContext;
@@ -188,7 +197,8 @@ public PolarisResolvedPathWrapper getPassthroughResolvedPath(Object key) {
 
     // Run a single-use Resolver for this path.
     Resolver passthroughResolver =
-        resolverFactory.createResolver(callContext, securityContext, catalogName);
+        resolverFactory.createResolver(
+            realmContext, realmConfig, metaStoreManager, securityContext, catalogName);
     passthroughResolver.addPath(requestedPath);
     ResolverStatus status = passthroughResolver.resolveAll();
 
@@ -273,7 +283,7 @@ public Set<PolarisBaseEntity> getAllActivatedPrincipalRoleEntities() {
     if (resolvedEntity == null) {
       LOGGER.warn(
           "Failed to find rootContainer for realm: {} and catalog: {}",
-          callContext.getRealmContext().getRealmIdentifier(),
+          realmContext.getRealmIdentifier(),
           catalogName);
     }
     return resolvedEntity;

polaris-core/src/main/java/org/apache/polaris/core/context/CallContext.java

@@ -22,13 +22,17 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import jakarta.ws.rs.core.SecurityContext;
-import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 
 public interface ResolutionManifestFactory {
 
   @Nonnull
   PolarisResolutionManifest createResolutionManifest(
-      @Nonnull CallContext callContext,
+      @Nonnull RealmContext realmContext,
+      @Nonnull RealmConfig realmConfig,
+      @Nonnull PolarisMetaStoreManager metaStoreManager,
       @Nonnull SecurityContext securityContext,
       @Nullable String referenceCatalogName);
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/PolarisResolutionManifest.java

@@ -23,7 +23,9 @@
 import jakarta.annotation.Nullable;
 import jakarta.ws.rs.core.SecurityContext;
 import org.apache.polaris.core.PolarisDiagnostics;
-import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 
 public class ResolutionManifestFactoryImpl implements ResolutionManifestFactory {
 
@@ -39,10 +41,18 @@ public ResolutionManifestFactoryImpl(
   @Nonnull
   @Override
   public PolarisResolutionManifest createResolutionManifest(
-      @Nonnull CallContext callContext,
+      @Nonnull RealmContext realmContext,
+      @Nonnull RealmConfig realmConfig,
+      @Nonnull PolarisMetaStoreManager metaStoreManager,
       @Nonnull SecurityContext securityContext,
       @Nullable String referenceCatalogName) {
     return new PolarisResolutionManifest(
-        diagnostics, callContext, resolverFactory, securityContext, referenceCatalogName);
+        diagnostics,
+        realmContext,
+        realmConfig,
+        metaStoreManager,
+        resolverFactory,
+        securityContext,
+        referenceCatalogName);
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/ResolutionManifestFactory.java

@@ -22,11 +22,15 @@
 import jakarta.annotation.Nonnull;
 import jakarta.annotation.Nullable;
 import jakarta.ws.rs.core.SecurityContext;
-import org.apache.polaris.core.context.CallContext;
+import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 
 public interface ResolverFactory {
   Resolver createResolver(
-      @Nonnull CallContext callContext,
+      @Nonnull RealmContext realmContext,
+      @Nonnull RealmConfig realmConfig,
+      @Nonnull PolarisMetaStoreManager metaStoreManager,
       @Nonnull SecurityContext securityContext,
       @Nullable String referenceCatalogName);
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/resolver/ResolutionManifestFactoryImpl.java

@@ -30,10 +30,10 @@
 import java.util.Set;
 import java.util.function.Function;
 import org.apache.iceberg.exceptions.UnprocessableEntityException;
-import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.config.FeatureConfiguration;
 import org.apache.polaris.core.config.RealmConfig;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.persistence.dao.entity.ScopedCredentialsResult;
@@ -96,7 +96,8 @@ private long maxCacheDurationMs(RealmConfig realmConfig) {
    * Either get from the cache or generate a new entry for a scoped creds
    *
    * @param credentialVendor the credential vendor used to generate a new scoped creds if needed
-   * @param callCtx the call context
+   * @param realmContext the realm context
+   * @param realmConfig the realm config
    * @param polarisEntity the polaris entity that is going to scoped creds
    * @param allowListOperation whether allow list action on the provided read and write locations
    * @param allowedReadLocations a set of allowed to read locations
@@ -105,7 +106,8 @@ private long maxCacheDurationMs(RealmConfig realmConfig) {
    */
   public AccessConfig getOrGenerateSubScopeCreds(
       @Nonnull PolarisCredentialVendor credentialVendor,
-      @Nonnull PolarisCallContext callCtx,
+      @Nonnull RealmContext realmContext,
+      @Nonnull RealmConfig realmConfig,
       @Nonnull PolarisEntity polarisEntity,
       boolean allowListOperation,
       @Nonnull Set<String> allowedReadLocations,
@@ -117,7 +119,7 @@ public AccessConfig getOrGenerateSubScopeCreds(
     }
     StorageCredentialCacheKey key =
         StorageCredentialCacheKey.of(
-            callCtx.getRealmContext().getRealmIdentifier(),
+            realmContext.getRealmIdentifier(),
             polarisEntity,
             allowListOperation,
             allowedReadLocations,
@@ -137,7 +139,7 @@ public AccessConfig getOrGenerateSubScopeCreds(
                   k.allowedWriteLocations(),
                   k.refreshCredentialsEndpoint());
           if (scopedCredentialsResult.isSuccess()) {
-            long maxCacheDurationMs = maxCacheDurationMs(callCtx.getRealmConfig());
+            long maxCacheDurationMs = maxCacheDurationMs(realmConfig);
             return new StorageCredentialCacheEntry(
                 scopedCredentialsResult.getAccessConfig(), maxCacheDurationMs);
           }