Make CallContextCatalogFactory request-scoped (#2972)

note the only non-test usage spot is `IcebergCatalogHandler#initializeCatalog`
and `IcebergCatalogHandler` is getting created by `IcebergCatalogAdapter`
which is already `@RequestScoped`.

Author: XN137

runtime/service/src/main/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandler.java

@@ -267,9 +267,7 @@ protected void initializeCatalog() {
       this.baseCatalog = federatedCatalog;
     } else {
       LOGGER.atInfo().log("Initializing non-federated catalog");
-      this.baseCatalog =
-          catalogFactory.createCallContextCatalog(
-              callContext, polarisPrincipal, resolutionManifest);
+      this.baseCatalog = catalogFactory.createCallContextCatalog(resolutionManifest);
     }
     this.namespaceCatalog =
         (baseCatalog instanceof SupportsNamespaces) ? (SupportsNamespaces) baseCatalog : null;

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

@@ -19,14 +19,9 @@
 package org.apache.polaris.service.context.catalog;
 
 import org.apache.iceberg.catalog.Catalog;
-import org.apache.polaris.core.auth.PolarisPrincipal;
-import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
 
 public interface CallContextCatalogFactory {
 
-  Catalog createCallContextCatalog(
-      CallContext context,
-      PolarisPrincipal polarisPrincipal,
-      PolarisResolutionManifest resolvedManifest);
+  Catalog createCallContextCatalog(PolarisResolutionManifest resolvedManifest);
 }

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

@@ -18,7 +18,7 @@
  */
 package org.apache.polaris.service.context.catalog;
 
-import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.context.RequestScoped;
 import jakarta.inject.Inject;
 import java.util.HashMap;
 import java.util.Map;
@@ -28,7 +28,7 @@
 import org.apache.polaris.core.auth.PolarisPrincipal;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.entity.CatalogEntity;
-import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
+import org.apache.polaris.core.persistence.PolarisMetaStoreManager;
 import org.apache.polaris.core.persistence.resolver.PolarisResolutionManifest;
 import org.apache.polaris.core.persistence.resolver.ResolverFactory;
 import org.apache.polaris.service.catalog.iceberg.IcebergCatalog;
@@ -39,7 +39,7 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-@ApplicationScoped
+@RequestScoped
 public class PolarisCallContextCatalogFactory implements CallContextCatalogFactory {
   private static final Logger LOGGER =
       LoggerFactory.getLogger(PolarisCallContextCatalogFactory.class);
@@ -49,47 +49,50 @@ public class PolarisCallContextCatalogFactory implements CallContextCatalogFacto
   private final AccessConfigProvider accessConfigProvider;
   private final FileIOFactory fileIOFactory;
   private final ResolverFactory resolverFactory;
-  private final MetaStoreManagerFactory metaStoreManagerFactory;
   private final PolarisEventListener polarisEventListener;
+  private final PolarisMetaStoreManager metaStoreManager;
+  private final CallContext callContext;
+  private final PolarisPrincipal principal;
 
   @Inject
   public PolarisCallContextCatalogFactory(
       PolarisDiagnostics diagnostics,
       ResolverFactory resolverFactory,
-      MetaStoreManagerFactory metaStoreManagerFactory,
       TaskExecutor taskExecutor,
       AccessConfigProvider accessConfigProvider,
       FileIOFactory fileIOFactory,
-      PolarisEventListener polarisEventListener) {
+      PolarisEventListener polarisEventListener,
+      PolarisMetaStoreManager metaStoreManager,
+      CallContext callContext,
+      PolarisPrincipal principal) {
     this.diagnostics = diagnostics;
     this.resolverFactory = resolverFactory;
-    this.metaStoreManagerFactory = metaStoreManagerFactory;
     this.taskExecutor = taskExecutor;
     this.accessConfigProvider = accessConfigProvider;
     this.fileIOFactory = fileIOFactory;
     this.polarisEventListener = polarisEventListener;
+    this.metaStoreManager = metaStoreManager;
+    this.callContext = callContext;
+    this.principal = principal;
   }
 
   @Override
-  public Catalog createCallContextCatalog(
-      CallContext context,
-      PolarisPrincipal polarisPrincipal,
-      final PolarisResolutionManifest resolvedManifest) {
+  public Catalog createCallContextCatalog(final PolarisResolutionManifest resolvedManifest) {
     CatalogEntity catalog = resolvedManifest.getResolvedCatalogEntity();
     String catalogName = catalog.getName();
 
-    String realm = context.getRealmContext().getRealmIdentifier();
+    String realm = callContext.getRealmContext().getRealmIdentifier();
     String catalogKey = realm + "/" + catalogName;
     LOGGER.debug("Initializing new BasePolarisCatalog for key: {}", catalogKey);
 
     IcebergCatalog catalogInstance =
         new IcebergCatalog(
             diagnostics,
             resolverFactory,
-            metaStoreManagerFactory.getOrCreateMetaStoreManager(context.getRealmContext()),
-            context,
+            metaStoreManager,
+            callContext,
             resolvedManifest,
-            polarisPrincipal,
+            principal,
             taskExecutor,
             accessConfigProvider,
             fileIOFactory,

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

@@ -257,6 +257,7 @@ public void before(TestInfo testInfo) {
     PrincipalEntity rootPrincipal =
         metaStoreManager.findRootPrincipal(polarisContext).orElseThrow();
     this.authenticatedRoot = PolarisPrincipal.of(rootPrincipal, Set.of());
+    QuarkusMock.installMockForType(authenticatedRoot, PolarisPrincipal.class);
 
     this.adminService =
         new PolarisAdminService(
@@ -518,36 +519,37 @@ public static class TestPolarisCallContextCatalogFactory
 
     @SuppressWarnings("unused") // Required by CDI
     protected TestPolarisCallContextCatalogFactory() {
-      this(null, null, null, null, null, null, null);
+      this(null, null, null, null, null, null, null, null, null);
     }
 
     @Inject
     public TestPolarisCallContextCatalogFactory(
         PolarisDiagnostics diagnostics,
         ResolverFactory resolverFactory,
-        MetaStoreManagerFactory metaStoreManagerFactory,
         TaskExecutor taskExecutor,
         AccessConfigProvider accessConfigProvider,
         FileIOFactory fileIOFactory,
-        PolarisEventListener polarisEventListener) {
+        PolarisEventListener polarisEventListener,
+        PolarisMetaStoreManager metaStoreManager,
+        CallContext callContext,
+        PolarisPrincipal principal) {
       super(
           diagnostics,
           resolverFactory,
-          metaStoreManagerFactory,
           taskExecutor,
           accessConfigProvider,
           fileIOFactory,
-          polarisEventListener);
+          polarisEventListener,
+          metaStoreManager,
+          callContext,
+          principal);
     }
 
     @Override
-    public Catalog createCallContextCatalog(
-        CallContext context,
-        PolarisPrincipal polarisPrincipal,
-        final PolarisResolutionManifest resolvedManifest) {
+    public Catalog createCallContextCatalog(PolarisResolutionManifest resolvedManifest) {
       // This depends on the BasePolarisCatalog allowing calling initialize multiple times
       // to override the previous config.
-      Catalog catalog = super.createCallContextCatalog(context, polarisPrincipal, resolvedManifest);
+      Catalog catalog = super.createCallContextCatalog(resolvedManifest);
       catalog.initialize(
           CATALOG_NAME,
           ImmutableMap.of(

runtime/service/src/test/java/org/apache/polaris/service/catalog/iceberg/IcebergCatalogHandlerAuthzTest.java

@@ -1124,8 +1124,7 @@ private IcebergCatalogHandler newWrapperWithFineGrainedAuthzDisabled() {
     CallContextCatalogFactory mockFactory = Mockito.mock(CallContextCatalogFactory.class);
 
     // Mock the catalog factory to return our regular catalog but with mocked config
-    Mockito.when(mockFactory.createCallContextCatalog(Mockito.any(), Mockito.any(), Mockito.any()))
-        .thenReturn(baseCatalog);
+    Mockito.when(mockFactory.createCallContextCatalog(Mockito.any())).thenReturn(baseCatalog);
 
     return newWrapperWithFineLevelAuthDisabled(Set.of(), CATALOG_NAME, mockFactory, false);
   }
@@ -1894,18 +1893,16 @@ public void testSendNotificationSufficientPrivileges() {
         new PolarisCallContextCatalogFactory(
             diagServices,
             resolverFactory,
-            managerFactory,
             Mockito.mock(),
             accessConfigProvider,
             new DefaultFileIOFactory(),
-            polarisEventListener) {
+            polarisEventListener,
+            metaStoreManager,
+            callContext,
+            authenticatedRoot) {
           @Override
-          public Catalog createCallContextCatalog(
-              CallContext context,
-              PolarisPrincipal polarisPrincipal,
-              PolarisResolutionManifest resolvedManifest) {
-            Catalog catalog =
-                super.createCallContextCatalog(context, polarisPrincipal, resolvedManifest);
+          public Catalog createCallContextCatalog(PolarisResolutionManifest resolvedManifest) {
+            Catalog catalog = super.createCallContextCatalog(resolvedManifest);
             String fileIoImpl = "org.apache.iceberg.inmemory.InMemoryFileIO";
             catalog.initialize(
                 externalCatalog, ImmutableMap.of(CatalogProperties.FILE_IO_IMPL, fileIoImpl));

runtime/service/src/testFixtures/java/org/apache/polaris/service/TestServices.java

@@ -206,6 +206,39 @@ public TestServices build() {
       PolarisMetaStoreManager metaStoreManager =
           metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
 
+      CreatePrincipalResult createdPrincipal =
+          metaStoreManager.createPrincipal(
+              callContext.getPolarisCallContext(),
+              new PrincipalEntity.Builder()
+                  .setName("test-principal")
+                  .setCreateTimestamp(Instant.now().toEpochMilli())
+                  .setCredentialRotationRequiredState()
+                  .build());
+      PolarisPrincipal principal = PolarisPrincipal.of(createdPrincipal.getPrincipal(), Set.of());
+
+      SecurityContext securityContext =
+          new SecurityContext() {
+            @Override
+            public Principal getUserPrincipal() {
+              return principal;
+            }
+
+            @Override
+            public boolean isUserInRole(String s) {
+              return false;
+            }
+
+            @Override
+            public boolean isSecure() {
+              return true;
+            }
+
+            @Override
+            public String getAuthenticationScheme() {
+              return "";
+            }
+          };
+
       EntityCache entityCache =
           metaStoreManagerFactory.getOrCreateEntityCache(realmContext, realmConfig);
       ResolverFactory resolverFactory =
@@ -251,11 +284,13 @@ public TestServices build() {
           new PolarisCallContextCatalogFactory(
               diagnostics,
               resolverFactory,
-              metaStoreManagerFactory,
               taskExecutor,
               accessConfigProvider,
               fileIOFactory,
-              polarisEventListener);
+              polarisEventListener,
+              metaStoreManager,
+              callContext,
+              principal);
 
       ReservedProperties reservedProperties = ReservedProperties.NONE;
 
@@ -289,39 +324,6 @@ public TestServices build() {
       IcebergRestConfigurationApi restConfigurationApi =
           new IcebergRestConfigurationApi(catalogService);
 
-      CreatePrincipalResult createdPrincipal =
-          metaStoreManager.createPrincipal(
-              callContext.getPolarisCallContext(),
-              new PrincipalEntity.Builder()
-                  .setName("test-principal")
-                  .setCreateTimestamp(Instant.now().toEpochMilli())
-                  .setCredentialRotationRequiredState()
-                  .build());
-      PolarisPrincipal principal = PolarisPrincipal.of(createdPrincipal.getPrincipal(), Set.of());
-
-      SecurityContext securityContext =
-          new SecurityContext() {
-            @Override
-            public Principal getUserPrincipal() {
-              return principal;
-            }
-
-            @Override
-            public boolean isUserInRole(String s) {
-              return false;
-            }
-
-            @Override
-            public boolean isSecure() {
-              return true;
-            }
-
-            @Override
-            public String getAuthenticationScheme() {
-              return "";
-            }
-          };
-
       PolarisAdminService adminService =
           new PolarisAdminService(
               callContext,