(polaris) branch main updated: Execute event listeners on a dedicated thread pool (#4616)

[ad...@apache.org]

This is an automated email from the ASF dual-hosted git repository.

adutra pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/polaris.git


The following commit(s) were added to refs/heads/main by this push:
     new ad6aec1b64 Execute event listeners on a dedicated thread pool (#4616)
ad6aec1b64 is described below

commit ad6aec1b649fb89443af06ef3aa9c27137af04b4
Author: Alexandre Dutra <ad...@apache.org>
AuthorDate: Thu Jun 4 19:29:17 2026 +0200

    Execute event listeners on a dedicated thread pool (#4616)
    
    This change introduces a new, separate thread pool for event listener execution; listeners now execute always on this executor's threads, instead of on the Vert.x event bus threads.
    
    Since event listeners are often blocking, because they execute I/O, running a blocking listener on the event bus thread should be avoided.
---
 CHANGELOG.md                                       |  4 ++-
 .../polaris/service/config/ServiceProducers.java   | 29 ++++++++++++++++++++++
 .../events/PolarisEventListenerConfiguration.java  | 18 ++++++++++++++
 .../service/events/PolarisEventListeners.java      | 24 ++++++++++++++++--
 .../service/events/PolarisEventListenersTest.java  |  3 +++
 .../smallrye-polaris_event_listener.md             |  2 ++
 6 files changed, 77 insertions(+), 3 deletions(-)

diff --git a/CHANGELOG.md b/CHANGELOG.md
index 904ad5dcf3..17c77b54b1 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -30,6 +30,9 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 ### Highlights
 
 ### Upgrade notes
+- Event listeners are now executed on a dedicated executor. **This executor does not propagate the original request's CDI context**; listeners that were improperly relying on that should instead manage their own CDI request scope from now on. Furthermore, two new configuration options were introduced to configure the executor: 
+  - `polaris.event-listener.executor.pool-size` configures the thread pool size.
+  - `polaris.event-listener.executor.queue-size` configures the queue size for pending events when all threads are busy.
 
 ### Breaking changes
 - The `MaintenanceService.performMaintenance()` signature now requires an explicit `OptionalLong overrideRunId` argument to supersede the latest unfinished maintenance run.
@@ -46,7 +49,6 @@ request adding CHANGELOG notes for breaking (!) changes and possibly other secti
 - Added documentation for BigQuery Metastore Catalog federation. Build with `-PNonRESTCatalogs=BIGQUERY` to include the BigQueryMetastoreCatalog federation extension. See `site/content/in-dev/unreleased/federation/bigquery-metastore-federation.md`.
 - Support for view registration has been added.
 - Python client: added support for Python 3.14
-
 - Added support for `register table` overwrite semantics in the Iceberg REST catalog flow (`overwrite=true`) for internal Polaris catalogs. With overwrite enabled, existing table pointers can be updated to a new metadata location while preserving default behavior for `overwrite=false`.
 - Added `REGISTER_TABLE_OVERWRITE` authorization operation mapped to `TABLE_FULL_METADATA` for deterministic overwrite authorization.
 
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java b/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
index d9b97f07f6..cec98cf4dc 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/config/ServiceProducers.java
@@ -32,6 +32,8 @@ import jakarta.enterprise.inject.Produces;
 import jakarta.inject.Singleton;
 import java.time.Clock;
 import java.util.Optional;
+import java.util.concurrent.Executor;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.stream.Collectors;
 import org.apache.polaris.core.PolarisCallContext;
 import org.apache.polaris.core.PolarisDefaultDiagServiceImpl;
@@ -74,6 +76,7 @@ import org.apache.polaris.service.catalog.io.FileIOFactory;
 import org.apache.polaris.service.context.RealmContextConfiguration;
 import org.apache.polaris.service.context.RealmContextResolver;
 import org.apache.polaris.service.credentials.PolarisCredentialManagerConfiguration;
+import org.apache.polaris.service.events.PolarisEventListenerConfiguration;
 import org.apache.polaris.service.persistence.PersistenceConfiguration;
 import org.apache.polaris.service.ratelimiter.RateLimiter;
 import org.apache.polaris.service.ratelimiter.RateLimiterFilterConfiguration;
@@ -459,4 +462,30 @@ public class ServiceProducers {
       MetricsReportingConfiguration config, @Any Instance<PolarisMetricsReporter> reporters) {
     return reporters.select(Identifier.Literal.of(config.type())).get();
   }
+
+  @Produces
+  @Singleton
+  @Identifier("event-listener-executor")
+  public Executor eventListenerExecutor(PolarisEventListenerConfiguration config) {
+    if (config.types().isEmpty()) {
+      return runnable -> {
+        throw new RejectedExecutionException(
+            "Event listener executor is not available because no event listeners are configured");
+      };
+    }
+    return SmallRyeManagedExecutor.builder()
+        .injectionPointName("event-listener-executor")
+        .propagated(ThreadContext.ALL_REMAINING)
+        .cleared(ThreadContext.CDI)
+        .maxAsync(config.executor().poolSize())
+        .maxQueued(config.executor().queueSize())
+        .build();
+  }
+
+  public void closeEventListenerExecutor(
+      @Disposes @Identifier("event-listener-executor") Executor executor) {
+    if (executor instanceof ManagedExecutor managedExecutor) {
+      managedExecutor.close();
+    }
+  }
 }
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
index 1924afb2bb..6ad16d7171 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListenerConfiguration.java
@@ -20,6 +20,7 @@ package org.apache.polaris.service.events;
 
 import io.quarkus.runtime.annotations.StaticInitSafe;
 import io.smallrye.config.ConfigMapping;
+import io.smallrye.config.WithDefault;
 import io.smallrye.config.WithParentName;
 import java.util.Map;
 import java.util.Optional;
@@ -67,4 +68,21 @@ public interface PolarisEventListenerConfiguration {
      */
     Optional<Set<PolarisEventType.Category>> enabledEventCategories();
   }
+
+  /** Configuration for the thread pool running event listeners. */
+  Executor executor();
+
+  interface Executor {
+
+    /**
+     * The thread pool size. The default is -1, which is interpreted as the number of available
+     * cores.
+     */
+    @WithDefault("-1")
+    int poolSize();
+
+    /** The queue size. The default is -1, which is interpreted as unbounded. */
+    @WithDefault("-1")
+    int queueSize();
+  }
 }
diff --git a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java
index b12cebc27e..870fd8ff6b 100644
--- a/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java
+++ b/runtime/service/src/main/java/org/apache/polaris/service/events/PolarisEventListeners.java
@@ -33,6 +33,7 @@ import jakarta.enterprise.inject.Instance;
 import jakarta.inject.Inject;
 import java.util.EnumSet;
 import java.util.HashSet;
+import java.util.concurrent.Executor;
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -45,6 +46,10 @@ public class PolarisEventListeners {
   @Inject @Any Instance<PolarisEventListener> eventListeners;
   @Inject PolarisEventListenerConfiguration configuration;
 
+  @Inject
+  @Identifier("event-listener-executor")
+  Executor executor;
+
   private final EnumSet<PolarisEventType> enabledEventTypes = EnumSet.allOf(PolarisEventType.class);
 
   public void onStartup(@Observes StartupEvent event) {
@@ -81,7 +86,23 @@ public class PolarisEventListeners {
       PolarisEvent event, String listenerName, PolarisEventListener listener) {
     LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
     try {
-      listener.onEvent(event);
+      executor.execute(
+          () -> {
+            LOGGER.debug(
+                "Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);
+            try {
+              listener.onEvent(event);
+            } catch (Exception e) {
+              LOGGER.error(
+                  "Error while delivering {} event to listener '{}' ({})",
+                  event.type(),
+                  listenerName,
+                  listener,
+                  e);
+            }
+            LOGGER.debug(
+                "Delivered {} event to listener '{}' ({})", event.type(), listenerName, listener);
+          });
     } catch (Exception e) {
       LOGGER.error(
           "Error while delivering {} event to listener '{}' ({})",
@@ -90,7 +111,6 @@ public class PolarisEventListeners {
           listener,
           e);
     }
-    LOGGER.debug("Delivered {} event to listener '{}' ({})", event.type(), listenerName, listener);
   }
 
   public boolean hasListeners(PolarisEventType polarisEventType) {
diff --git a/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersTest.java b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersTest.java
index 93fd1346b6..39553e9f8b 100644
--- a/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersTest.java
+++ b/runtime/service/src/test/java/org/apache/polaris/service/events/PolarisEventListenersTest.java
@@ -19,6 +19,7 @@
 
 package org.apache.polaris.service.events;
 
+import static org.assertj.core.api.Assertions.assertThat;
 import static org.awaitility.Awaitility.await;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -28,6 +29,7 @@ import io.quarkus.test.junit.QuarkusTest;
 import io.quarkus.test.junit.QuarkusTestProfile;
 import io.quarkus.test.junit.TestProfile;
 import io.smallrye.common.annotation.Identifier;
+import io.vertx.core.Context;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import java.util.HashSet;
@@ -61,6 +63,7 @@ public class PolarisEventListenersTest {
 
     @Override
     public void onEvent(PolarisEvent event) {
+      assertThat(Context.isOnEventLoopThread()).isFalse();
       if (predicate.test(event)) {
         expectedEvents.add(event);
       } else {
diff --git a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
index 4d0a544550..c19a7088dc 100644
--- a/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
+++ b/site/content/in-dev/unreleased/configuration/config-sections/smallrye-polaris_event_listener.md
@@ -29,3 +29,5 @@ build:
 | `polaris.event-listener.types` |  | `list of string` | Comma separated list of event listeners, each item must be a registered (`PolarisEventListener`) identifier.   |
 | `polaris.event-listener.`_`<name>`_`.enabled-event-types` |  | <span title="BEFORE_CREATE_CATALOG, AFTER_CREATE_CATALOG, BEFORE_DELETE_CATALOG, AFTER_DELETE_CATALOG, BEFORE_GET_CATALOG, AFTER_GET_CATALOG, BEFORE_UPDATE_CATALOG, AFTER_UPDATE_CATALOG, BEFORE_LIST_CATALOGS, AFTER_LIST_CATALOGS, BEFORE_CREATE_CATALOG_ROLE, AFTER_CREATE_CATALOG_ROLE, BEFORE_DELETE_CATALOG_ROLE, AFTER_DELETE_CATALOG_ROLE, BEFORE_GET_CATALOG_ROLE, AFTER_GET_CATALOG_ROLE, BEFORE_UPDATE_CATALOG_ROLE, AFTER_UPDA [...]
 | `polaris.event-listener.`_`<name>`_`.enabled-event-categories` |  | <span title="CATALOG, NAMESPACE, TABLE, VIEW, POLICY, GENERIC_TABLE, PRINCIPAL_ROLE, PRINCIPAL, CATALOG_ROLE, CREDENTIAL, TRANSACTION, NOTIFICATION, CONFIG, TASK_EXECUTION, RATE_LIMITING"><code>list of enum (CATALOG, NAMESPACE, TABLE, ...)</code></span> | Comma separated list of enabled event type categories. Each category is a collection of  related Polaris event types. This event listener will only receive events of  [...]
+| `polaris.event-listener.executor.pool-size` | `-1` | `int` | The thread pool size. The default is -1, which is interpreted as the number of available  cores.  |
+| `polaris.event-listener.executor.queue-size` | `-1` | `int` | The queue size. The default is -1, which is interpreted as unbounded. |