Avoid metrics persistence setup for in-memory event buffering

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

@@ -40,13 +40,22 @@
  */
 public class PolarisCallContext implements CallContext {
 
+  private static final MetricsPersistence NO_OP_METRICS_PERSISTENCE = new MetricsPersistence() {};
+
   // 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;
 
+  private static MetricsPersistence metricsPersistenceOrNoOp(BasePersistence metaStore) {
+    if (metaStore instanceof MetricsPersistence metricsPersistence) {
+      return metricsPersistence;
+    }
+    return NO_OP_METRICS_PERSISTENCE;
+  }
+
   /**
    * @deprecated Use {@link PolarisCallContext#PolarisCallContext(RealmContext, BasePersistence,
    *     MetricsPersistence, RealmConfigurationSource)}.
@@ -96,12 +105,13 @@ public PolarisCallContext(
   }
 
   /**
-   * Convenience constructor for callers whose persistence type satisfies both SPIs and who do not
-   * have a {@link RealmConfigurationSource}.
+   * Convenience constructor for callers that do not need to supply distinct metrics persistence.
+   * Backends whose metastore also implements {@link MetricsPersistence} use the metastore for both
+   * SPIs; other backends use no-op metrics persistence.
    */
-  public <P extends BasePersistence & MetricsPersistence> PolarisCallContext(
-      @NonNull RealmContext realmContext, @NonNull P metaStore) {
-    this(realmContext, metaStore, metaStore, RealmConfigurationSource.EMPTY_CONFIG);
+  public PolarisCallContext(
+      @NonNull RealmContext realmContext, @NonNull BasePersistence metaStore) {
+    this(realmContext, metaStore, metricsPersistenceOrNoOp(metaStore));
   }
 
   public BasePersistence getMetaStore() {

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

@@ -168,8 +168,7 @@ protected void flush(String realmId, List<EventEntity> 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);
   }
 

runtime/service/src/test/java/org/apache/polaris/service/events/listeners/inmemory/InMemoryBufferEventListenerBufferSizeTest.java

@@ -24,6 +24,8 @@
 import static org.mockito.ArgumentMatchers.any;
 import static org.mockito.Mockito.doReturn;
 import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
 
 import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.TestProfile;
@@ -46,6 +48,7 @@ void testFlushOnSize() {
     sendAsync("test2", 10);
     assertRows("test1", 10);
     assertRows("test2", 10);
+    verify(metaStoreManagerFactory, never()).getOrCreateMetricsPersistence(any());
   }
 
   @Test