Inject MetricsPersistence directly into PersistingMetricsReporter

persistence/nosql/persistence/metastore-maintenance/src/test/java/org/apache/polaris/persistence/nosql/metastore/maintenance/TestCatalogMaintenance.java

@@ -569,8 +569,7 @@ private TestSetup bootstrapRealm() {
 
     var manager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
     var session = metaStoreManagerFactory.getOrCreateSession(realmContext);
-    var metrics = metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
-    var callCtx = new PolarisCallContext(realmContext, session, metrics, configurationSource);
+    var callCtx = new PolarisCallContext(realmContext, session, configurationSource);
     var persistence =
         realmPersistenceFactory.newBuilder().realmId(realmId).skipDecorators().build();
 
@@ -927,8 +926,7 @@ private void checkEntities(String step, List<PolarisBaseEntity> entities) {
 
     var manager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
     var session = metaStoreManagerFactory.getOrCreateSession(realmContext);
-    var metrics = metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
-    var callCtx = new PolarisCallContext(realmContext, session, metrics, configurationSource);
+    var callCtx = new PolarisCallContext(realmContext, session, configurationSource);
 
     for (var e : entities) {
       var result =

persistence/nosql/persistence/metastore/src/main/java/org/apache/polaris/persistence/nosql/metastore/NoSqlMetaStoreManagerFactory.java

@@ -283,11 +283,7 @@ private PrincipalSecretsResult bootstrapRealm(
             clock);
 
     PolarisCallContext ctx =
-        new PolarisCallContext(
-            () -> realmId,
-            metaStore,
-            NO_OP_METRICS_PERSISTENCE,
-            RealmConfigurationSource.EMPTY_CONFIG);
+        new PolarisCallContext(() -> realmId, metaStore, RealmConfigurationSource.EMPTY_CONFIG);
     var secretsResult = createPolarisPrincipalForRealm(metaStoreManager, ctx);
 
     realmManagement.update(

persistence/nosql/persistence/metastore/src/test/java/org/apache/polaris/persistence/nosql/metastore/TestNoSqlMetaStoreManager.java

@@ -105,9 +105,8 @@ protected PolarisTestMetaStoreManager createPolarisTestMetaStoreManager() {
 
     var manager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
     var session = metaStoreManagerFactory.getOrCreateSession(realmContext);
-    var metrics = metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
 
-    var callCtx = new PolarisCallContext(realmContext, session, metrics, configurationSource);
+    var callCtx = new PolarisCallContext(realmContext, session, configurationSource);
 
     return new PolarisTestMetaStoreManager(manager, callCtx, startTime, false);
   }

persistence/nosql/persistence/metastore/src/test/java/org/apache/polaris/persistence/nosql/metastore/TestNoSqlResolver.java

@@ -72,8 +72,7 @@ protected PolarisCallContext callCtx() {
 
       metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
       var session = metaStoreManagerFactory.getOrCreateSession(realmContext);
-      var metrics = metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
-      callCtx = new PolarisCallContext(realmContext, session, metrics, configurationSource);
+      callCtx = new PolarisCallContext(realmContext, session, configurationSource);
 
       tm = new PolarisTestMetaStoreManager(metaStoreManager, callCtx, startTime, false);
     }

persistence/nosql/persistence/metastore/src/test/java/org/apache/polaris/persistence/nosql/metastore/indexaccess/TestIndexedContainerAccess.java

@@ -97,9 +97,7 @@ public void setup(TestInfo testInfo) {
     var realmContext = (RealmContext) () -> realmId;
     callContext =
         new PolarisCallContext(
-            realmContext,
-            metaStoreManagerFactory.getOrCreateSession(realmContext),
-            metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext));
+            realmContext, metaStoreManagerFactory.getOrCreateSession(realmContext));
     metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
   }
 

persistence/nosql/persistence/metastore/src/test/java/org/apache/polaris/persistence/nosql/metastore/indexaccess/TestMemoizedIndexAccess.java

@@ -92,9 +92,7 @@ public void setup(TestInfo testInfo) {
     var realmContext = (RealmContext) () -> realmId;
     callContext =
         new PolarisCallContext(
-            realmContext,
-            metaStoreManagerFactory.getOrCreateSession(realmContext),
-            metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext));
+            realmContext, metaStoreManagerFactory.getOrCreateSession(realmContext));
     metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
 
     memoized = MemoizedIndexedAccess.newMemoizedIndexedAccess(persistence);

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

@@ -25,93 +25,59 @@
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.context.RealmContext;
 import org.apache.polaris.core.persistence.BasePersistence;
-import org.apache.polaris.core.persistence.metrics.MetricsPersistence;
 import org.jspecify.annotations.NonNull;
 
 /**
  * The Call context is allocated each time a new REST request is processed. It contains instances of
  * low-level services required to process that request.
  *
- * <p>{@link BasePersistence} continues to carry the bulk of the metastore SPI surface (and still
- * extends {@code PolicyMappingPersistence} / acts as the {@code IntegrationPersistence} via a
- * runtime cast for now). {@link MetricsPersistence} is intentionally decoupled and supplied
- * separately so callers that only need metrics persistence do not have to depend on {@link
- * BasePersistence}.
+ * <p>{@link BasePersistence} carries the bulk of the metastore SPI surface (and still extends
+ * {@code PolicyMappingPersistence} / acts as the {@code IntegrationPersistence} via a runtime cast
+ * for now). {@link org.apache.polaris.core.persistence.metrics.MetricsPersistence} is intentionally
+ * kept out of this context; service code that needs metrics persistence should depend on that SPI
+ * directly.
  */
 public class PolarisCallContext implements CallContext {
 
   // meta store which is used to persist Polaris entity metadata
   private final BasePersistence metaStore;
-  private final MetricsPersistence metricsPersistence;
   private final RealmConfigurationSource configurationSource;
   private final RealmContext realmContext;
   private final RealmConfig realmConfig;
 
   /**
    * @deprecated Use {@link PolarisCallContext#PolarisCallContext(RealmContext, BasePersistence,
-   *     MetricsPersistence, RealmConfigurationSource)}.
+   *     RealmConfigurationSource)}.
    */
   @SuppressWarnings("removal")
   @Deprecated(forRemoval = true)
   public PolarisCallContext(
       @NonNull RealmContext realmContext,
       @NonNull BasePersistence metaStore,
       @NonNull PolarisConfigurationStore configurationStore) {
-    this(
-        realmContext, metaStore, new MetricsPersistence() {}, configurationStore::getConfiguration);
+    this(realmContext, metaStore, configurationStore::getConfiguration);
   }
 
-  /**
-   * Convenience constructor for backends whose {@link BasePersistence} implementation also
-   * implements {@link MetricsPersistence} (the common in-tree case). Callers that need to wire a
-   * distinct metrics implementation should use {@link #PolarisCallContext(RealmContext,
-   * BasePersistence, MetricsPersistence, RealmConfigurationSource)}.
-   */
-  public <P extends BasePersistence & MetricsPersistence> PolarisCallContext(
-      @NonNull RealmContext realmContext,
-      @NonNull P metaStore,
-      @NonNull RealmConfigurationSource configurationSource) {
-    this(realmContext, metaStore, metaStore, configurationSource);
-  }
-
-  /** Primary constructor — {@link MetricsPersistence} is supplied separately from the metastore. */
   public PolarisCallContext(
       @NonNull RealmContext realmContext,
       @NonNull BasePersistence metaStore,
-      @NonNull MetricsPersistence metricsPersistence,
       @NonNull RealmConfigurationSource configurationSource) {
     this.realmContext = realmContext;
     this.metaStore = metaStore;
-    this.metricsPersistence = metricsPersistence;
     this.configurationSource = configurationSource;
     this.realmConfig = new RealmConfigImpl(this.configurationSource, this.realmContext);
   }
 
   /** Convenience constructor that defaults to {@link RealmConfigurationSource#EMPTY_CONFIG}. */
   public PolarisCallContext(
-      @NonNull RealmContext realmContext,
-      @NonNull BasePersistence metaStore,
-      @NonNull MetricsPersistence metricsPersistence) {
-    this(realmContext, metaStore, metricsPersistence, RealmConfigurationSource.EMPTY_CONFIG);
-  }
-
-  /**
-   * Convenience constructor for callers whose persistence type satisfies both SPIs and who do not
-   * have a {@link RealmConfigurationSource}.
-   */
-  public <P extends BasePersistence & MetricsPersistence> PolarisCallContext(
-      @NonNull RealmContext realmContext, @NonNull P metaStore) {
-    this(realmContext, metaStore, metaStore, RealmConfigurationSource.EMPTY_CONFIG);
+      @NonNull RealmContext realmContext, @NonNull BasePersistence metaStore) {
+    this(realmContext, metaStore, RealmConfigurationSource.EMPTY_CONFIG);
   }
 
   public BasePersistence getMetaStore() {
     return metaStore;
   }
 
-  public MetricsPersistence getMetricsPersistence() {
-    return metricsPersistence;
-  }
-
   @Override
   public RealmContext getRealmContext() {
     return realmContext;
@@ -136,7 +102,6 @@ public PolarisCallContext copy() {
     // copy of the RealmContext to ensure the access during the task executor.
     String realmId = this.realmContext.getRealmIdentifier();
     RealmContext realmContext = () -> realmId;
-    return new PolarisCallContext(
-        realmContext, this.metaStore, this.metricsPersistence, this.configurationSource);
+    return new PolarisCallContext(realmContext, this.metaStore, this.configurationSource);
   }
 }

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

@@ -48,7 +48,6 @@
 import org.apache.polaris.core.persistence.dao.entity.ListEntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.ResolvedEntitiesResult;
 import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
-import org.apache.polaris.core.persistence.metrics.PolarisMetricsManager;
 import org.apache.polaris.core.persistence.pagination.Page;
 import org.apache.polaris.core.persistence.pagination.PageToken;
 import org.apache.polaris.core.policy.PolarisPolicyMappingManager;
@@ -63,8 +62,7 @@ public interface PolarisMetaStoreManager
     extends PolarisSecretsManager,
         PolarisGrantManager,
         PolarisPolicyMappingManager,
-        PolarisEventManager,
-        PolarisMetricsManager {
+        PolarisEventManager {
 
   /**
    * Bootstrap the Polaris service, creating the root catalog, root principal, and associated

polaris-core/src/main/java/org/apache/polaris/core/persistence/metrics/MetricsPersistence.java

@@ -42,8 +42,6 @@
  *
  * <p><b>Note:</b> This SPI is currently experimental and not yet implemented in all persistence
  * backends. The API may change in future releases.
- *
- * @see PolarisMetricsManager
  */
 @Beta
 public interface MetricsPersistence {

polaris-core/src/test/java/org/apache/polaris/core/persistence/AtomicOperationMetaStoreManagerRefreshTest.java

@@ -36,7 +36,6 @@
 import org.apache.polaris.core.entity.PolarisEntityType;
 import org.apache.polaris.core.persistence.dao.entity.BaseResult;
 import org.apache.polaris.core.persistence.dao.entity.ResolvedEntityResult;
-import org.apache.polaris.core.persistence.metrics.MetricsPersistence;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import org.mockito.Mockito;
@@ -58,7 +57,7 @@ public class AtomicOperationMetaStoreManagerRefreshTest {
   public void setUp() {
     diagnostics = new PolarisDefaultDiagServiceImpl();
     metaStore = Mockito.mock(BasePersistence.class);
-    callCtx = new PolarisCallContext(() -> "testRealm", metaStore, new MetricsPersistence() {});
+    callCtx = new PolarisCallContext(() -> "testRealm", metaStore);
     manager = new AtomicOperationMetaStoreManager(Clock.systemUTC(), diagnostics);
   }
 

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

@@ -140,13 +140,14 @@ public CallContext polarisCallContext(
       RealmConfigurationSource configurationSource,
       MetaStoreManagerFactory metaStoreManagerFactory) {
     BasePersistence metaStore = metaStoreManagerFactory.getOrCreateSession(realmContext);
-    // When the backend implements both SPIs on the same instance (e.g. JDBC, in-memory), reuse the
-    // session instead of building a second persistence instance per request.
-    MetricsPersistence metricsPersistence =
-        (metaStore instanceof MetricsPersistence mp)
-            ? mp
-            : metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
-    return new PolarisCallContext(realmContext, metaStore, metricsPersistence, configurationSource);
+    return new PolarisCallContext(realmContext, metaStore, configurationSource);
+  }
+
+  @Produces
+  @RequestScoped
+  public MetricsPersistence metricsPersistence(
+      RealmContext realmContext, MetaStoreManagerFactory metaStoreManagerFactory) {
+    return metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
   }
 
   @Produces

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

@@ -109,8 +109,7 @@ protected void flush(String realmId, List<PolarisEvent> events) {
     RealmContext realmContext = () -> realmId;
     var metaStoreManager = metaStoreManagerFactory.getOrCreateMetaStoreManager(realmContext);
     var basePersistence = metaStoreManagerFactory.getOrCreateSession(realmContext);
-    var metricsPersistence = metaStoreManagerFactory.getOrCreateMetricsPersistence(realmContext);
-    var callContext = new PolarisCallContext(realmContext, basePersistence, metricsPersistence);
+    var callContext = new PolarisCallContext(realmContext, basePersistence);
     metaStoreManager.writeEvents(callContext, events);
   }