[fix][broker] Skip loading broker interceptor when disableBrokerInterceptors is true

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

@@ -1347,7 +1347,7 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
 
     @FieldContext(
         category = CATEGORY_SERVER,
-        doc = "Enable or disable the broker interceptor, which is only used for testing for now"
+        doc = "Enable or disable the broker interceptor"
     )
     private boolean disableBrokerInterceptors = true;
 

pulsar-broker/src/main/java/org/apache/pulsar/broker/intercept/BrokerInterceptors.java

@@ -59,6 +59,11 @@ public BrokerInterceptors(Map<String, BrokerInterceptorWithClassLoader> intercep
      * @return the collection of broker event interceptor
      */
     public static BrokerInterceptor load(ServiceConfiguration conf) throws IOException {
+        if (conf.isDisableBrokerInterceptors()) {
+            log.info("Skip loading the broker interceptors when disableBrokerInterceptors is true");
+            return null;
+        }
+
         BrokerInterceptorDefinitions definitions =
                 BrokerInterceptorUtils.searchForInterceptors(conf.getBrokerInterceptorsDirectory(),
                         conf.getNarExtractionDirectory());

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

@@ -20,9 +20,12 @@
 
 import static org.mockito.ArgumentMatchers.same;
 import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.times;
 import static org.mockito.Mockito.verify;
 import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNull;
+
 import java.io.IOException;
 import java.util.HashMap;
 import java.util.HashSet;
@@ -36,6 +39,7 @@
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
+import org.apache.pulsar.broker.ServiceConfiguration;
 import org.apache.pulsar.broker.testcontext.PulsarTestContext;
 import org.apache.pulsar.client.admin.PulsarAdminException;
 import org.apache.pulsar.client.api.Consumer;
@@ -307,4 +311,25 @@ public void requestInterceptorFailedTest() {
         }
     }
 
+    @Test
+    public void testLoadWhenDisableBrokerInterceptorsIsTrue() throws IOException {
+        ServiceConfiguration serviceConfiguration = spy(ServiceConfiguration.class);
+        serviceConfiguration.setDisableBrokerInterceptors(true);
+        BrokerInterceptor brokerInterceptor = BrokerInterceptors.load(serviceConfiguration);
+        assertNull(brokerInterceptor);
+
+        verify(serviceConfiguration, times(1)).isDisableBrokerInterceptors();
+        verify(serviceConfiguration, times(0)).getBrokerInterceptorsDirectory();
+    }
+
+    @Test
+    public void testLoadWhenDisableBrokerInterceptorsIsFalse() throws IOException {
+        ServiceConfiguration serviceConfiguration = spy(ServiceConfiguration.class);
+        serviceConfiguration.setDisableBrokerInterceptors(false);
+        BrokerInterceptor brokerInterceptor = BrokerInterceptors.load(serviceConfiguration);
+        assertNull(brokerInterceptor);
+
+        verify(serviceConfiguration, times(1)).isDisableBrokerInterceptors();
+        verify(serviceConfiguration, times(1)).getBrokerInterceptorsDirectory();
+    }
 }