Event Listeners

quarkus/defaults/src/main/resources/application.properties

@@ -124,6 +124,8 @@ polaris.secrets-manager.type=in-memory
 
 polaris.file-io.type=default
 
+polaris.event-listener.type=no-op
+
 polaris.log.request-id-header-name=Polaris-Request-Id
 # polaris.log.mdc.aid=polaris
 # polaris.log.mdc.sid=polaris-service

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/config/ProductionReadinessChecks.java

@@ -36,6 +36,8 @@
 import org.apache.polaris.service.context.DefaultRealmContextResolver;
 import org.apache.polaris.service.context.RealmContextResolver;
 import org.apache.polaris.service.context.TestRealmContextResolver;
+import org.apache.polaris.service.events.PolarisEventListener;
+import org.apache.polaris.service.events.TestPolarisEventListener;
 import org.apache.polaris.service.persistence.InMemoryPolarisMetaStoreManagerFactory;
 import org.apache.polaris.service.quarkus.auth.QuarkusAuthenticationConfiguration;
 import org.eclipse.microprofile.config.Config;
@@ -161,6 +163,16 @@ public ProductionReadinessCheck checkRealmResolver(Config config, RealmContextRe
     return ProductionReadinessCheck.OK;
   }
 
+  @Produces
+  public ProductionReadinessCheck checkPolarisEventListener(
+      PolarisEventListener polarisEventListener) {
+    if (polarisEventListener instanceof TestPolarisEventListener) {
+      return ProductionReadinessCheck.of(
+          Error.of("TestPolarisEventListener is intended for tests only.", "polaris.events.type"));
+    }
+    return ProductionReadinessCheck.OK;
+  }
+
   private static String authRealmSegment(String realm) {
     return realm.equals(QuarkusAuthenticationConfiguration.DEFAULT_REALM_KEY) ? "" : realm + ".";
   }

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/config/QuarkusProducers.java

@@ -61,11 +61,13 @@
 import org.apache.polaris.service.context.RealmContextConfiguration;
 import org.apache.polaris.service.context.RealmContextFilter;
 import org.apache.polaris.service.context.RealmContextResolver;
+import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.quarkus.auth.QuarkusAuthenticationConfiguration;
 import org.apache.polaris.service.quarkus.auth.QuarkusAuthenticationRealmConfiguration;
 import org.apache.polaris.service.quarkus.auth.external.tenant.OidcTenantResolver;
 import org.apache.polaris.service.quarkus.catalog.io.QuarkusFileIOConfiguration;
 import org.apache.polaris.service.quarkus.context.QuarkusRealmContextConfiguration;
+import org.apache.polaris.service.quarkus.events.QuarkusPolarisEventListenerConfiguration;
 import org.apache.polaris.service.quarkus.persistence.QuarkusPersistenceConfiguration;
 import org.apache.polaris.service.quarkus.ratelimiter.QuarkusRateLimiterFilterConfiguration;
 import org.apache.polaris.service.quarkus.ratelimiter.QuarkusTokenBucketConfiguration;
@@ -151,6 +153,13 @@ public FileIOFactory fileIOFactory(
     return fileIOFactories.select(Identifier.Literal.of(config.type())).get();
   }
 
+  @Produces
+  public PolarisEventListener polarisEventListener(
+      QuarkusPolarisEventListenerConfiguration config,
+      @Any Instance<PolarisEventListener> polarisEventListeners) {
+    return polarisEventListeners.select(Identifier.Literal.of(config.type())).get();
+  }
+
   @Produces
   public MetaStoreManagerFactory metaStoreManagerFactory(
       QuarkusPersistenceConfiguration config,

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/events/QuarkusPolarisEventListenerConfiguration.java

@@ -0,0 +1,32 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.polaris.service.quarkus.events;
+
+import io.quarkus.runtime.annotations.StaticInitSafe;
+import io.smallrye.config.ConfigMapping;
+
+@StaticInitSafe
+@ConfigMapping(prefix = "polaris.event-listener")
+public interface QuarkusPolarisEventListenerConfiguration {
+  /**
+   * The type of the event listener to use. Must be a registered {@link
+   * org.apache.polaris.service.events.PolarisEventListener} identifier.
+   */
+  String type();
+}

quarkus/service/src/main/java/org/apache/polaris/service/quarkus/task/QuarkusTaskExecutorImpl.java

@@ -29,6 +29,7 @@
 import java.util.concurrent.ExecutorService;
 import org.apache.polaris.core.context.CallContext;
 import org.apache.polaris.core.persistence.MetaStoreManagerFactory;
+import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.quarkus.tracing.QuarkusTracingFilter;
 import org.apache.polaris.service.task.TaskExecutorImpl;
 import org.apache.polaris.service.task.TaskFileIOSupplier;
@@ -39,16 +40,17 @@ public class QuarkusTaskExecutorImpl extends TaskExecutorImpl {
   private final Tracer tracer;
 
   public QuarkusTaskExecutorImpl() {
-    this(null, null, null, null);
+    this(null, null, null, null, null);
   }
 
   @Inject
   public QuarkusTaskExecutorImpl(
       @Identifier("task-executor") ExecutorService executorService,
       MetaStoreManagerFactory metaStoreManagerFactory,
       TaskFileIOSupplier fileIOSupplier,
-      Tracer tracer) {
-    super(executorService, metaStoreManagerFactory, fileIOSupplier);
+      Tracer tracer,
+      PolarisEventListener polarisEventListener) {
+    super(executorService, metaStoreManagerFactory, fileIOSupplier, polarisEventListener);
     this.tracer = tracer;
   }
 

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

@@ -88,6 +88,7 @@
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
 import org.apache.polaris.service.context.CallContextCatalogFactory;
 import org.apache.polaris.service.context.PolarisCallContextCatalogFactory;
+import org.apache.polaris.service.events.PolarisEventListener;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
 import org.apache.polaris.service.task.TaskExecutor;
 import org.apache.polaris.service.types.PolicyIdentifier;
@@ -188,6 +189,7 @@ public Map<String, String> getConfigOverrides() {
   @Inject protected PolarisDiagnostics diagServices;
   @Inject protected Clock clock;
   @Inject protected FileIOFactory fileIOFactory;
+  @Inject protected PolarisEventListener polarisEventListener;
 
   protected IcebergCatalog baseCatalog;
   protected GenericTableCatalog genericTableCatalog;
@@ -469,7 +471,8 @@ private void initBaseCatalog() {
             passthroughView,
             securityContext,
             Mockito.mock(),
-            fileIOFactory);
+            fileIOFactory,
+            polarisEventListener);
     this.baseCatalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(
@@ -485,7 +488,7 @@ public static class TestPolarisCallContextCatalogFactory
       extends PolarisCallContextCatalogFactory {
 
     public TestPolarisCallContextCatalogFactory() {
-      super(null, null, null, null, null);
+      super(null, null, null, null, null, null);
     }
 
     @Inject
@@ -494,13 +497,15 @@ public TestPolarisCallContextCatalogFactory(
         MetaStoreManagerFactory metaStoreManagerFactory,
         UserSecretsManagerFactory userSecretsManagerFactory,
         TaskExecutor taskExecutor,
-        FileIOFactory fileIOFactory) {
+        FileIOFactory fileIOFactory,
+        PolarisEventListener polarisEventListener) {
       super(
           entityManagerFactory,
           metaStoreManagerFactory,
           userSecretsManagerFactory,
           taskExecutor,
-          fileIOFactory);
+          fileIOFactory,
+          polarisEventListener);
     }
 
     @Override

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/GenericTableCatalogTest.java

@@ -81,6 +81,7 @@
 import org.apache.polaris.service.catalog.io.DefaultFileIOFactory;
 import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.config.RealmEntityManagerFactory;
+import org.apache.polaris.service.events.NoOpPolarisEventListener;
 import org.apache.polaris.service.storage.PolarisStorageIntegrationProviderImpl;
 import org.apache.polaris.service.task.TaskExecutor;
 import org.assertj.core.api.Assertions;
@@ -264,7 +265,8 @@ public void before(TestInfo testInfo) {
             passthroughView,
             securityContext,
             taskExecutor,
-            fileIOFactory);
+            fileIOFactory,
+            new NoOpPolarisEventListener());
     this.icebergCatalog.initialize(
         CATALOG_NAME,
         ImmutableMap.of(

quarkus/service/src/test/java/org/apache/polaris/service/quarkus/catalog/IcebergCatalogHandlerAuthzTest.java

@@ -1808,7 +1808,8 @@ public PolarisEntityManager getOrCreateEntityManager(RealmContext realmContext)
             userSecretsManagerFactory,
             Mockito.mock(),
             new DefaultFileIOFactory(
-                realmEntityManagerFactory, managerFactory, new PolarisConfigurationStore() {})) {
+                realmEntityManagerFactory, managerFactory, new PolarisConfigurationStore() {}),
+            polarisEventListener) {
           @Override
           public Catalog createCallContextCatalog(
               CallContext context,