Enable intercept filters only when interceptors are configured

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -750,13 +750,20 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private Set<String> brokerInterceptors = Sets.newTreeSet();
 
+    @FieldContext(
+        category = CATEGORY_SERVER,
+        doc = "Enable or disable the broker interceptor, which is only used for testing for now"
+    )
+    private boolean disableBrokerInterceptors = true;
+
     @FieldContext(
         doc = "There are two policies when zookeeper session expired happens, \"shutdown\" and \"reconnect\". \n\n"
         + " If uses \"shutdown\" policy, shutdown the broker when zookeeper session expired happens.\n\n"
         + " If uses \"reconnect\" policy, try to reconnect to zookeeper server and re-register metadata to zookeeper."
     )
     private String zookeeperSessionExpiredPolicy = "shutdown";
 
+
     /**** --- Messaging Protocols --- ****/
 
     @FieldContext(

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/ResponseHandlerFilter.java

@@ -43,10 +43,12 @@ public class ResponseHandlerFilter implements Filter {
 
     private final String brokerAddress;
     private final BrokerInterceptor interceptor;
+    private final boolean interceptorEnabled;
 
     public ResponseHandlerFilter(PulsarService pulsar) {
         this.brokerAddress = pulsar.getAdvertisedAddress();
         this.interceptor = pulsar.getBrokerInterceptor();
+        this.interceptorEnabled = !pulsar.getConfig().getBrokerInterceptors().isEmpty();
     }
 
     @Override
@@ -63,8 +65,9 @@ public void doFilter(ServletRequest request, ServletResponse response, FilterCha
                 /* connection is already invalidated */
             }
         }
-        interceptor.onWebserviceResponse(request, response);
-
+        if (interceptorEnabled) {
+            interceptor.onWebserviceResponse(request, response);
+        }
     }
 
     @Override

pulsar-broker/src/main/java/org/apache/pulsar/broker/web/WebService.java

@@ -155,8 +155,11 @@ public void addServlet(String path, ServletHolder servletHolder, boolean require
             });
         }
 
-        context.addFilter(new FilterHolder(new PreInterceptFilter(pulsar.getBrokerInterceptor())),
+        if (!pulsar.getConfig().getBrokerInterceptors().isEmpty() || !pulsar.getConfig().isDisableBrokerInterceptors()) {
+            // Enable PreInterceptFilter only when interceptors are enabled
+            context.addFilter(new FilterHolder(new PreInterceptFilter(pulsar.getBrokerInterceptor())),
                 MATCH_ALL, EnumSet.allOf(DispatcherType.class));
+        }
 
         if (requiresAuthentication && pulsar.getConfiguration().isAuthenticationEnabled()) {
             FilterHolder filter = new FilterHolder(new AuthenticationFilter(

pulsar-broker/src/test/java/org/apache/pulsar/broker/intercept/BrokerInterceptorTest.java

@@ -51,6 +51,8 @@ public class BrokerInterceptorTest extends ProducerConsumerBase {
 
     @BeforeMethod
     public void setup() throws Exception {
+        this.conf.setDisableBrokerInterceptors(false);
+
         this.listener1 = mock(BrokerInterceptor.class);
         this.ncl1 = mock(NarClassLoader.class);
         this.listener2 = mock(BrokerInterceptor.class);