(polaris) branch main updated: Refactor event listener supported type resolution (#4625)

[yu...@apache.org]

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

flyrain 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 bacb5f149f Refactor event listener supported type resolution (#4625)
bacb5f149f is described below

commit bacb5f149f3b4e7b5b3b706cbceeac7b57cdffe9
Author: Yufei Gu <yu...@apache.org>
AuthorDate: Tue Jun 9 11:32:28 2026 -0700

    Refactor event listener supported type resolution (#4625)
---
 .../service/events/PolarisEventListeners.java      | 47 ++++++++++++++--------
 1 file changed, 31 insertions(+), 16 deletions(-)

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 26118f0557..35924744a2 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.Set;
 import java.util.concurrent.Executor;
 import org.apache.polaris.service.events.listeners.PolarisEventListener;
 import org.slf4j.Logger;
@@ -53,27 +54,19 @@ public class PolarisEventListeners {
   private final EnumSet<PolarisEventType> eventTypesWithListeners =
       EnumSet.noneOf(PolarisEventType.class);
 
+  /**
+   * Registers configured Polaris event listeners with the local Vert.x event bus at startup.
+   *
+   * <p>Each listener may enable event categories, individual event types, or both. When no
+   * per-listener filters are configured, the listener receives every Polaris event type.
+   */
   public void onStartup(@Observes StartupEvent event) {
     var listenerTypeSet = configuration.types().orElseGet(HashSet::new);
     for (String enabledEventListener : listenerTypeSet) {
       var listenerConfiguration = configuration.listenerConfig().get(enabledEventListener);
-      var supportedTypes =
-          listenerConfiguration == null
-                  || (listenerConfiguration.enabledEventCategories().isEmpty()
-                      && listenerConfiguration.enabledEventTypes().isEmpty())
-              ? EnumSet.allOf(PolarisEventType.class)
-              : EnumSet.noneOf(PolarisEventType.class);
-      if (listenerConfiguration != null) {
-        if (listenerConfiguration.enabledEventCategories().isPresent()) {
-          for (var enabledEventCategory : listenerConfiguration.enabledEventCategories().get()) {
-            supportedTypes.addAll(PolarisEventType.typesOfCategory(enabledEventCategory));
-          }
-        }
-        if (listenerConfiguration.enabledEventTypes().isPresent()) {
-          supportedTypes.addAll(listenerConfiguration.enabledEventTypes().get());
-        }
-      }
+      var supportedTypes = resolveSupportedTypes(listenerConfiguration);
       var listener = eventListeners.select(Identifier.Literal.of(enabledEventListener)).get();
+      // Reuse the same handler for every selected event type for this listener.
       Handler<Message<PolarisEvent>> handler =
           message -> deliverEvent(message.body(), enabledEventListener, listener);
       for (var polarisEventType : supportedTypes) {
@@ -83,6 +76,28 @@ public class PolarisEventListeners {
     }
   }
 
+  private static EnumSet<PolarisEventType> resolveSupportedTypes(
+      PolarisEventListenerConfiguration.ListenerConfiguration listenerConfiguration) {
+    // Missing or empty per-listener filters mean that the listener receives all event types.
+    if (listenerConfiguration == null
+        || (listenerConfiguration.enabledEventCategories().orElse(Set.of()).isEmpty()
+            && listenerConfiguration.enabledEventTypes().orElse(Set.of()).isEmpty())) {
+      return EnumSet.allOf(PolarisEventType.class);
+    }
+
+    // Category and type filters are additive: a listener receives the union of both.
+    var supportedTypes = EnumSet.noneOf(PolarisEventType.class);
+    if (listenerConfiguration.enabledEventCategories().isPresent()) {
+      for (var enabledEventCategory : listenerConfiguration.enabledEventCategories().get()) {
+        supportedTypes.addAll(PolarisEventType.typesOfCategory(enabledEventCategory));
+      }
+    }
+    if (listenerConfiguration.enabledEventTypes().isPresent()) {
+      supportedTypes.addAll(listenerConfiguration.enabledEventTypes().get());
+    }
+    return supportedTypes;
+  }
+
   private void deliverEvent(
       PolarisEvent event, String listenerName, PolarisEventListener listener) {
     LOGGER.debug("Delivering {} event to listener '{}' ({})", event.type(), listenerName, listener);