use CallContext in MetaStoreManagerFactory.createMetaStoreManager

Author: XN137

persistence/relational-jdbc/src/main/java/org/apache/polaris/persistence/relational/jdbc/JdbcMetaStoreManagerFactory.java

@@ -35,6 +35,7 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.config.RealmConfigImpl;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.AtomicOperationMetaStoreManager;
@@ -200,7 +201,11 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
   }
 
   @Override
-  public PolarisMetaStoreManager createMetaStoreManager(RealmContext realmContext) {
+  public PolarisMetaStoreManager createMetaStoreManager(CallContext callContext) {
+    return createMetaStoreManager(callContext.getRealmContext());
+  }
+
+  private PolarisMetaStoreManager createMetaStoreManager(RealmContext realmContext) {
     RealmConfig realmConfig = new RealmConfigImpl(configurationStore, realmContext);
     return new AtomicOperationMetaStoreManager(
         clock,

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

@@ -29,6 +29,7 @@
 import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.config.RealmConfig;
 import org.apache.polaris.core.config.RealmConfigImpl;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PrincipalEntity;
 import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet;
@@ -149,7 +150,11 @@ public Map<String, BaseResult> purgeRealms(Iterable<String> realms) {
   }
 
   @Override
-  public PolarisMetaStoreManager createMetaStoreManager(RealmContext realmContext) {
+  public PolarisMetaStoreManager createMetaStoreManager(CallContext callContext) {
+    return createMetaStoreManager(callContext.getRealmContext());
+  }
+
+  private PolarisMetaStoreManager createMetaStoreManager(RealmContext realmContext) {
     RealmConfig realmConfig = new RealmConfigImpl(configurationStore, realmContext);
     return createNewMetaStoreManager(clock, diagnostics, realmContext, realmConfig);
   }

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

@@ -19,6 +19,7 @@
 package org.apache.polaris.core.persistence;
 
 import java.util.Map;
+import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.bootstrap.BootstrapOptions;
 import org.apache.polaris.core.persistence.bootstrap.RootCredentialsSet;
@@ -29,7 +30,7 @@
 /** Configuration interface for configuring the {@link PolarisMetaStoreManager}. */
 public interface MetaStoreManagerFactory {
 
-  PolarisMetaStoreManager createMetaStoreManager(RealmContext realmContext);
+  PolarisMetaStoreManager createMetaStoreManager(CallContext callContext);
 
   EntityCache getOrCreateEntityCache(RealmContext realmContext);
 

runtime/service/src/main/java/org/apache/polaris/service/catalog/io/DefaultFileIOFactory.java

@@ -31,7 +31,6 @@
 import org.apache.iceberg.catalog.TableIdentifier;
 import org.apache.iceberg.io.FileIO;
 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.persistence.MetaStoreManagerFactory;
 import org.apache.polaris.core.persistence.PolarisResolvedPathWrapper;
@@ -72,9 +71,8 @@ public FileIO loadFileIO(
       @Nonnull Set<String> tableLocations,
       @Nonnull Set<PolarisStorageActions> storageActions,
       @Nonnull PolarisResolvedPathWrapper resolvedEntityPath) {
-    RealmContext realmContext = callContext.getRealmContext();
     PolarisCredentialVendor credentialVendor =
-        metaStoreManagerFactory.createMetaStoreManager(realmContext);
+        metaStoreManagerFactory.createMetaStoreManager(callContext);
 
     // Get subcoped creds
     properties = new HashMap<>(properties);

runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java

@@ -196,8 +196,8 @@ public MetaStoreManagerFactory metaStoreManagerFactory(
   @Produces
   @RequestScoped
   public PolarisMetaStoreManager polarisMetaStoreManager(
-      MetaStoreManagerFactory metaStoreManagerFactory, RealmContext realmContext) {
-    return metaStoreManagerFactory.createMetaStoreManager(realmContext);
+      MetaStoreManagerFactory metaStoreManagerFactory, CallContext callContext) {
+    return metaStoreManagerFactory.createMetaStoreManager(callContext);
   }
 
   @Produces

runtime/service/src/main/java/org/apache/polaris/service/context/catalog/PolarisCallContextCatalogFactory.java

@@ -82,7 +82,7 @@ public Catalog createCallContextCatalog(
     LOGGER.debug("Initializing new BasePolarisCatalog for key: {}", catalogKey);
 
     PolarisMetaStoreManager metaStoreManager =
-        metaStoreManagerFactory.createMetaStoreManager(context.getRealmContext());
+        metaStoreManagerFactory.createMetaStoreManager(context);
     IcebergCatalog catalogInstance =
         new IcebergCatalog(
             diagnostics,

runtime/service/src/main/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListener.java

@@ -39,6 +39,8 @@
 import java.time.Duration;
 import java.util.List;
 import java.util.Objects;
+import org.apache.polaris.core.PolarisCallContext;
+import org.apache.polaris.core.config.PolarisConfigurationStore;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.entity.PolarisEvent;
@@ -57,6 +59,7 @@ public class InMemoryBufferEventListener extends PolarisPersistenceEventListener
 
   @Inject CallContext callContext;
   @Inject Clock clock;
+  @Inject PolarisConfigurationStore configurationStore;
   @Inject MetaStoreManagerFactory metaStoreManagerFactory;
   @Inject InMemoryBufferEventListenerConfiguration configuration;
 
@@ -118,7 +121,8 @@ protected UnicastProcessor<PolarisEvent> createProcessor(String realmId) {
   @Fallback(fallbackMethod = "onFlushError")
   protected void flush(String realmId, List<PolarisEvent> events) {
     RealmContext realmContext = () -> realmId;
-    var metaStoreManager = metaStoreManagerFactory.createMetaStoreManager(realmContext);
+    CallContext callContext = new PolarisCallContext(realmContext, configurationStore);
+    var metaStoreManager = metaStoreManagerFactory.createMetaStoreManager(callContext);
     metaStoreManager.writeEvents(events);
   }
 

runtime/service/src/main/java/org/apache/polaris/service/task/TableCleanupTaskHandler.java

@@ -107,7 +107,7 @@ public boolean handleTask(TaskEntity cleanupTask, CallContext callContext) {
           TableMetadataParser.read(fileIO, tableEntity.getMetadataLocation());
 
       PolarisMetaStoreManager metaStoreManager =
-          metaStoreManagerFactory.createMetaStoreManager(callContext.getRealmContext());
+          metaStoreManagerFactory.createMetaStoreManager(callContext);
 
       Stream<TaskEntity> manifestCleanupTasks =
           getManifestTaskStream(cleanupTask, tableMetadata, fileIO, tableEntity, metaStoreManager);

runtime/service/src/main/java/org/apache/polaris/service/task/TaskExecutorImpl.java

@@ -149,7 +149,7 @@ protected void handleTask(long taskEntityId, CallContext ctx, int attempt) {
     try {
       LOGGER.info("Handling task entity id {}", taskEntityId);
       PolarisMetaStoreManager metaStoreManager =
-          metaStoreManagerFactory.createMetaStoreManager(ctx.getRealmContext());
+          metaStoreManagerFactory.createMetaStoreManager(ctx);
       PolarisBaseEntity taskEntity =
           metaStoreManager.loadEntity(0L, taskEntityId, PolarisEntityType.TASK).getEntity();
       if (!PolarisEntityType.TASK.equals(taskEntity.getType())) {

runtime/service/src/test/java/org/apache/polaris/service/admin/PolarisAuthzTestBase.java

@@ -251,7 +251,7 @@ public void before(TestInfo testInfo) {
 
     callContext = polarisContext;
     realmConfig = polarisContext.getRealmConfig();
-    metaStoreManager = managerFactory.createMetaStoreManager(realmContext);
+    metaStoreManager = managerFactory.createMetaStoreManager(polarisContext);
     userSecretsManager = userSecretsManagerFactory.getOrCreateUserSecretsManager(realmContext);
 
     polarisAuthorizer = new PolarisAuthorizerImpl(realmConfig);