Add filter system topic when using EventCenter (#1219)

streamnative · Jan 23, 2024 · 7e4ea20 · 7e4ea20
1 parent 9efcb3e
commit 7e4ea20
Show file tree

Hide file tree

Showing 2 changed files with 35 additions and 18 deletions.
diff --git a/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/event/PulsarEventCenterImpl.java b/...c/main/java/io/streamnative/pulsar/handlers/mqtt/support/event/PulsarEventCenterImpl.java
@@ -19,6 +19,7 @@
 import java.util.stream.Collectors;
 import lombok.extern.slf4j.Slf4j;
 import org.apache.bookkeeper.common.util.OrderedExecutor;
+import org.apache.commons.collections4.CollectionUtils;
 import org.apache.pulsar.broker.service.BrokerService;
 import org.apache.pulsar.metadata.api.Notification;
 
@@ -59,25 +60,28 @@ public void accept(Notification notification) {
             return;
         }
         final String path = notification.getPath();
-        // give the task to another thread to avoid blocking metadata
-        callbackExecutor.executeOrdered(path, () -> {
-            for (PulsarEventListener listener : listeners.stream().filter(listeners ->
-                    listeners.matchPath(path)).collect(Collectors.toList())) {
-                try {
-                    switch (notification.getType()) {
-                        case Created:
-                            listener.onNodeCreated(path);
-                            break;
-                        case Deleted:
-                            listener.onNodeDeleted(path);
-                            break;
-                        default:
-                            break;
+        final List<PulsarEventListener> matcherListeners = listeners.stream().filter(listeners ->
+                listeners.matchPath(path)).collect(Collectors.toList());
+        if (CollectionUtils.isNotEmpty(matcherListeners)) {
+            // give the task to another thread to avoid blocking metadata
+            callbackExecutor.executeOrdered(path, () -> {
+                for (PulsarEventListener listener : matcherListeners) {
+                    try {
+                        switch (notification.getType()) {
+                            case Created:
+                                listener.onNodeCreated(path);
+                                break;
+                            case Deleted:
+                                listener.onNodeDeleted(path);
+                                break;
+                            default:
+                                break;
+                        }
+                    } catch (Throwable ex) {
+                        log.warn("notify change {} {} failed.", notification.getType(), notification.getPath(), ex);
                     }
-                } catch (Throwable ex) {
-                    log.warn("notify change {} {} failed.", notification.getType(), notification.getPath(), ex);
                 }
-            }
-        });
+            });
+        }
     }
 }
diff --git a/...in/java/io/streamnative/pulsar/handlers/mqtt/support/event/PulsarTopicChangeListener.java b/...in/java/io/streamnative/pulsar/handlers/mqtt/support/event/PulsarTopicChangeListener.java
@@ -16,13 +16,16 @@
 import io.streamnative.pulsar.handlers.mqtt.utils.EventParserUtils;
 import java.util.regex.Pattern;
 import org.apache.commons.lang.StringUtils;
+import org.apache.pulsar.common.naming.SystemTopicNames;
 import org.apache.pulsar.common.naming.TopicName;
 
 public interface PulsarTopicChangeListener extends PulsarEventListener {
     String MANAGED_LEDGER_PATH = "/managed-ledgers";
     String REGEX = MANAGED_LEDGER_PATH + "/" + ".*" + "/persistent/";
     Pattern PATTERN = Pattern.compile(REGEX);
 
+    boolean FILTER_SYSTEM_TOPIC = true;
+
     @Override
     default boolean matchPath(String path) {
         String listenNamespace = getListenNamespace();
@@ -41,6 +44,9 @@ default String getListenNamespace() {
     default void onNodeCreated(String path) {
         try {
             TopicName topicName = EventParserUtils.parseTopicNameFromManagedLedgerEvent(path);
+            if (isFilterSystemTopic() && SystemTopicNames.isSystemTopic(topicName)) {
+                return;
+            }
             onTopicLoad(topicName);
         } catch (IllegalArgumentException ex) {
             // NO-OP don't notify
@@ -51,6 +57,9 @@ default void onNodeCreated(String path) {
     default void onNodeDeleted(String path) {
         try {
             TopicName topicName = EventParserUtils.parseTopicNameFromManagedLedgerEvent(path);
+            if (isFilterSystemTopic() && SystemTopicNames.isSystemTopic(topicName)) {
+                return;
+            }
             onTopicUnload(topicName);
         } catch (IllegalArgumentException ex) {
             // NO-OP don't notify
@@ -60,4 +69,8 @@ default void onNodeDeleted(String path) {
     void onTopicLoad(TopicName topicName);
 
     void onTopicUnload(TopicName topicName);
+
+    default boolean isFilterSystemTopic() {
+        return FILTER_SYSTEM_TOPIC;
+    }
 }