[fix][broker] Disable EntryFilters for system topics (#20514)

Author: lhotari

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractTopic.java

@@ -1163,6 +1163,10 @@ public void updateResourceGroupLimiter(Optional<Policies> optPolicies) {
     }
 
     public void updateEntryFilters() {
+        if (isSystemTopic()) {
+            entryFilters = Pair.of(null, Collections.emptyList());
+            return;
+        }
         final EntryFilters entryFiltersPolicy = getEntryFiltersPolicy();
         if (entryFiltersPolicy == null || StringUtils.isBlank(entryFiltersPolicy.getEntryFilterNames())) {
             entryFilters = Pair.of(null, Collections.emptyList());

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryFilterSupport.java

@@ -35,7 +35,8 @@ public class EntryFilterSupport {
 
     public EntryFilterSupport(Subscription subscription) {
         this.subscription = subscription;
-        if (subscription != null && subscription.getTopic() != null) {
+        if (subscription != null && subscription.getTopic() != null
+                && !subscription.getTopic().isSystemTopic()) {
             final BrokerService brokerService = subscription.getTopic().getBrokerService();
             final boolean allowOverrideEntryFilters = brokerService
                     .pulsar().getConfiguration().isAllowOverrideEntryFilters();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/SystemTopic.java

@@ -18,13 +18,16 @@
  */
 package org.apache.pulsar.broker.service.persistent;
 
+import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.mledger.ManagedLedger;
 import org.apache.pulsar.broker.PulsarServerException;
 import org.apache.pulsar.broker.namespace.NamespaceService;
 import org.apache.pulsar.broker.service.BrokerService;
+import org.apache.pulsar.broker.service.plugin.EntryFilter;
 import org.apache.pulsar.common.naming.SystemTopicNames;
 import org.apache.pulsar.common.naming.TopicName;
+import org.apache.pulsar.common.policies.data.EntryFilters;
 
 public class SystemTopic extends PersistentTopic {
 
@@ -82,4 +85,14 @@ public boolean isEncryptionRequired() {
         // System topics are only written by the broker that can't know the encryption context.
         return false;
     }
+
+    @Override
+    public EntryFilters getEntryFiltersPolicy() {
+        return null;
+    }
+
+    @Override
+    public List<EntryFilter> getEntryFilters() {
+        return null;
+    }
 }