Remove CallContext

extension/persistence/eclipselink/src/test/java/org/apache/polaris/extension/persistence/impl/eclipselink/PolarisEclipseLinkMetaStoreManagerTest.java

@@ -38,6 +38,7 @@
 import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisPrincipalSecrets;
 import org.apache.polaris.core.persistence.BasePolarisMetaStoreManagerTest;
 import org.apache.polaris.core.persistence.PolarisMetaStoreManagerImpl;
@@ -100,11 +101,13 @@ static void deleteConfFiles() throws IOException {
   protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
     PolarisDiagnostics diagServices = new PolarisDefaultDiagServiceImpl();
     PolarisEclipseLinkStore store = new PolarisEclipseLinkStore(diagServices);
+    RealmContext realmContext = () -> "realm";
     PolarisMetaStoreSession session =
         new PolarisEclipseLinkMetaStoreSessionImpl(
-            store, Mockito.mock(), () -> "realm", null, "polaris", RANDOM_SECRETS, diagServices);
+            store, Mockito.mock(), realmContext, null, "polaris", RANDOM_SECRETS, diagServices);
     return new PolarisTestMetaStoreManager(
         new PolarisMetaStoreManagerImpl(
+            realmContext,
             diagServices,
             new PolarisConfigurationStore() {},
             timeSource.withZone(ZoneId.systemDefault())),

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

@@ -23,6 +23,7 @@
 import jakarta.annotation.Nullable;
 import java.util.ArrayList;
 import java.util.List;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.CatalogEntity;
 
 /**
@@ -33,26 +34,29 @@ public interface PolarisConfigurationStore {
   /**
    * Retrieve the current value for a configuration key. May be null if not set.
    *
+   * @param <T> the type of the configuration value
+   * @param realmContext
    * @param configName the name of the configuration key to check
    * @return the current value set for the configuration key or null if not set
-   * @param <T> the type of the configuration value
    */
-  default <T> @Nullable T getConfiguration(String configName) {
+  default <T> @Nullable T getConfiguration(RealmContext realmContext, String configName) {
     return null;
   }
 
   /**
    * Retrieve the current value for a configuration key. If not set, return the non-null default
    * value.
    *
+   * @param <T> the type of the configuration value
+   * @param realmContext
    * @param configName the name of the configuration key to check
    * @param defaultValue the default value if the configuration key has no value
    * @return the current value or the supplied default value
-   * @param <T> the type of the configuration value
    */
-  default <T> @Nonnull T getConfiguration(String configName, @Nonnull T defaultValue) {
+  default <T> @Nonnull T getConfiguration(
+      RealmContext realmContext, String configName, @Nonnull T defaultValue) {
     Preconditions.checkNotNull(defaultValue, "Cannot pass null as a default value");
-    T configValue = getConfiguration(configName);
+    T configValue = getConfiguration(realmContext, configName);
     return configValue != null ? configValue : defaultValue;
   }
 
@@ -83,31 +87,36 @@ public interface PolarisConfigurationStore {
   /**
    * Retrieve the current value for a configuration.
    *
+   * @param <T> the type of the configuration value
+   * @param realmContext
    * @param config the configuration to load
    * @return the current value set for the configuration key or null if not set
-   * @param <T> the type of the configuration value
    */
-  default <T> @Nonnull T getConfiguration(PolarisConfiguration<T> config) {
-    T result = getConfiguration(config.key, config.defaultValue);
+  default <T> @Nonnull T getConfiguration(
+      RealmContext realmContext, PolarisConfiguration<T> config) {
+    T result = getConfiguration(realmContext, config.key, config.defaultValue);
     return tryCast(config, result);
   }
 
   /**
    * Retrieve the current value for a configuration, overriding with a catalog config if it is
    * present.
    *
+   * @param <T> the type of the configuration value
+   * @param realmContext
    * @param catalogEntity the catalog to check for an override
    * @param config the configuration to load
    * @return the current value set for the configuration key or null if not set
-   * @param <T> the type of the configuration value
    */
   default <T> @Nonnull T getConfiguration(
-      @Nonnull CatalogEntity catalogEntity, PolarisConfiguration<T> config) {
+      RealmContext realmContext,
+      @Nonnull CatalogEntity catalogEntity,
+      PolarisConfiguration<T> config) {
     if (config.hasCatalogConfig()
         && catalogEntity.getPropertiesAsMap().containsKey(config.catalogConfig())) {
       return tryCast(config, catalogEntity.getPropertiesAsMap().get(config.catalogConfig()));
     } else {
-      return getConfiguration(config);
+      return getConfiguration(realmContext, config);
     }
   }
 }

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizer.java

@@ -22,20 +22,23 @@
 import jakarta.annotation.Nullable;
 import java.util.List;
 import java.util.Set;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
 
 /** Interface for invoking authorization checks. */
 public interface PolarisAuthorizer {
 
   void authorizeOrThrow(
+      @Nonnull RealmContext realmContext,
       @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable PolarisResolvedPathWrapper target,
       @Nullable PolarisResolvedPathWrapper secondary);
 
   void authorizeOrThrow(
+      @Nonnull RealmContext realmContext,
       @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,

polaris-core/src/main/java/org/apache/polaris/core/auth/PolarisAuthorizerImpl.java

@@ -100,6 +100,7 @@
 import org.apache.iceberg.exceptions.ForbiddenException;
 import org.apache.polaris.core.PolarisConfiguration;
 import org.apache.polaris.core.PolarisConfigurationStore;
+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.PolarisEntityCore;
@@ -486,12 +487,14 @@ public boolean matchesOrIsSubsumedBy(
 
   @Override
   public void authorizeOrThrow(
+      @Nonnull RealmContext realmContext,
       @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable PolarisResolvedPathWrapper target,
       @Nullable PolarisResolvedPathWrapper secondary) {
     authorizeOrThrow(
+        realmContext,
         authenticatedPrincipal,
         activatedEntities,
         authzOp,
@@ -501,13 +504,15 @@ public void authorizeOrThrow(
 
   @Override
   public void authorizeOrThrow(
+      @Nonnull RealmContext realmContext,
       @Nonnull AuthenticatedPolarisPrincipal authenticatedPrincipal,
       @Nonnull Set<PolarisBaseEntity> activatedEntities,
       @Nonnull PolarisAuthorizableOperation authzOp,
       @Nullable List<PolarisResolvedPathWrapper> targets,
       @Nullable List<PolarisResolvedPathWrapper> secondaries) {
     boolean enforceCredentialRotationRequiredState =
         featureConfig.getConfiguration(
+            realmContext,
             PolarisConfiguration.ENFORCE_PRINCIPAL_CREDENTIAL_ROTATION_REQUIRED_CHECKING);
     if (enforceCredentialRotationRequiredState
         && authenticatedPrincipal

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

@@ -24,5 +24,11 @@
  * prod), and/or account.
  */
 public interface RealmContext {
+
+  static RealmContext copyOf(RealmContext original) {
+    String realmIdentifier = original.getRealmIdentifier();
+    return () -> realmIdentifier;
+  }
+
   String getRealmIdentifier();
 }

polaris-core/src/main/java/org/apache/polaris/core/persistence/LocalPolarisMetaStoreManagerFactory.java

@@ -27,7 +27,6 @@
 import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.PolarisDiagnostics;
 import org.apache.polaris.core.auth.PolarisSecretsManager.PrincipalSecretsResult;
-import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEntity;
 import org.apache.polaris.core.entity.PolarisEntityConstants;
@@ -90,7 +89,7 @@ private void initializeForRealm(RealmContext realmContext) {
         () -> createMetaStoreSession(backingStore, realmContext, diagnostics));
 
     PolarisMetaStoreManager metaStoreManager =
-        new PolarisMetaStoreManagerImpl(diagnostics, configurationStore, clock);
+        new PolarisMetaStoreManagerImpl(realmContext, diagnostics, configurationStore, clock);
     metaStoreManagerMap.put(realmContext.getRealmIdentifier(), metaStoreManager);
   }
 
@@ -189,7 +188,6 @@ private PrincipalSecretsResult bootstrapServiceAndCreatePolarisPrincipalForRealm
     // CallContext hasn't even been resolved yet.
     PolarisMetaStoreSession metaStoreSession =
         sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
-    CallContext.setCurrentContext(CallContext.of(realmContext));
 
     PolarisMetaStoreManager.EntityResult preliminaryRootPrincipalLookup =
         metaStoreManager.readEntityByName(
@@ -245,7 +243,6 @@ private void checkPolarisServiceBootstrappedForRealm(
 
     PolarisMetaStoreSession metaStoreSession =
         sessionSupplierMap.get(realmContext.getRealmIdentifier()).get();
-    CallContext.setCurrentContext(CallContext.of(realmContext));
 
     PolarisMetaStoreManager.EntityResult rootPrincipalLookup =
         metaStoreManager.readEntityByName(

polaris-core/src/main/java/org/apache/polaris/core/persistence/PolarisMetaStoreManagerImpl.java

@@ -38,6 +38,7 @@
 import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisConfigurationStore;
 import org.apache.polaris.core.PolarisDiagnostics;
+import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.AsyncTaskType;
 import org.apache.polaris.core.entity.PolarisBaseEntity;
 import org.apache.polaris.core.entity.PolarisChangeTrackingVersions;
@@ -75,12 +76,17 @@ public class PolarisMetaStoreManagerImpl implements PolarisMetaStoreManager {
   /** use synchronous drop for entities */
   private static final boolean USE_SYNCHRONOUS_DROP = true;
 
+  private final RealmContext realmContext;
   private final PolarisDiagnostics diagnostics;
   private final PolarisConfigurationStore configurationStore;
   private final Clock clock;
 
   public PolarisMetaStoreManagerImpl(
-      PolarisDiagnostics diagnostics, PolarisConfigurationStore configurationStore, Clock clock) {
+      RealmContext realmContext,
+      PolarisDiagnostics diagnostics,
+      PolarisConfigurationStore configurationStore,
+      Clock clock) {
+    this.realmContext = realmContext;
     this.diagnostics = diagnostics;
     this.configurationStore = configurationStore;
     this.clock = clock;
@@ -1809,6 +1815,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
                   PolarisObjectMapperUtil.parseTaskState(entity);
               long taskAgeTimeout =
                   configurationStore.getConfiguration(
+                      realmContext,
                       PolarisTaskConstants.TASK_TIMEOUT_MILLIS_CONFIG,
                       PolarisTaskConstants.TASK_TIMEOUT_MILLIS);
               return taskState == null
@@ -1927,7 +1934,7 @@ private PolarisEntityResolver resolveSecurableToRoleGrant(
         readStorageConfiguration(diagnostics, reloadedEntity.getEntity());
     Map<String, String> validateLocationAccess =
         storageIntegration
-            .validateAccessToLocations(storageConfigurationInfo, actions, locations)
+            .validateAccessToLocations(realmContext, storageConfigurationInfo, actions, locations)
             .entrySet()
             .stream()
             .collect(