Add broker interceptor for Intercepting all Pulsar command and REST API requests

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

@@ -91,6 +91,8 @@ public class ServiceConfiguration implements PulsarConfiguration {
     private static final String CATEGORY_HTTP = "HTTP";
     @Category
     private static final String CATEGORY_TRANSACTION = "Transaction";
+    @Category
+    private static final String CATEGORY_INTERCEPTORS = "Interceptors";
 
     /***** --- pulsar configuration --- ****/
     @FieldContext(
@@ -707,6 +709,18 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private int maxNumPartitionsPerPartitionedTopic = 0;
 
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "The directory to locate broker listeners"
+    )
+    private String brokerListenersDirectory = "./brokerListeners";
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "List of broker listener to load, which is a list of broker listener names"
+    )
+    private Set<String> brokerListeners = Sets.newTreeSet();
+
     @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"
@@ -1769,6 +1783,20 @@ public class ServiceConfiguration implements PulsarConfiguration {
     )
     private Set<String> brokerClientTlsProtocols = Sets.newTreeSet();
 
+    /**** --- Interceptor variables --- ****/
+
+    @FieldContext(
+        category = CATEGORY_INTERCEPTORS,
+        doc = "The directory to locate interceptors"
+    )
+    private String interceptorDirectory = "./interceptors";
+
+    @FieldContext(
+        category = CATEGORY_INTERCEPTORS,
+        doc = "List of interceptors to load, which is a list of interceptor names"
+    )
+    private Set<String> interceptors = Sets.newTreeSet();
+
     /**
      * @deprecated See {@link #getConfigurationStoreServers}
      */

pulsar-broker/src/main/java/org/apache/pulsar/broker/PulsarService.java

@@ -77,6 +77,8 @@
 import org.apache.pulsar.broker.authorization.AuthorizationService;
 import org.apache.pulsar.broker.cache.ConfigurationCacheService;
 import org.apache.pulsar.broker.cache.LocalZooKeeperCacheService;
+import org.apache.pulsar.broker.events.BrokerEventListener;
+import org.apache.pulsar.broker.events.BrokerEventListeners;
 import org.apache.pulsar.broker.loadbalance.LeaderElectionService;
 import org.apache.pulsar.broker.loadbalance.LeaderElectionService.LeaderListener;
 import org.apache.pulsar.broker.loadbalance.LoadManager;
@@ -201,6 +203,7 @@ public class PulsarService implements AutoCloseable {
 
     private MetricsGenerator metricsGenerator;
     private TransactionMetadataStoreService transactionMetadataStoreService;
+    private BrokerEventListener brokerEventListener;
 
     public enum State {
         Init, Started, Closed
@@ -449,7 +452,9 @@ public void start() throws PulsarServerException {
 
             this.defaultOffloader = createManagedLedgerOffloader(
                     OffloadPolicies.create(this.getConfiguration().getProperties()));
-
+            this.brokerEventListener = BrokerEventListeners.load(config);
+            brokerService.setEventListener(getBrokerEventListener());
+            this.brokerEventListener.initialize(config);
             brokerService.start();
 
             this.webService = new WebService(this);

pulsar-broker/src/main/java/org/apache/pulsar/broker/events/BrokerEventListener.java

@@ -0,0 +1,96 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.events;
+
+import com.google.common.annotations.Beta;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.service.ServerCnx;
+import org.apache.pulsar.common.api.proto.PulsarApi.BaseCommand;
+
+import javax.servlet.FilterChain;
+import javax.servlet.ServletException;
+import javax.servlet.ServletRequest;
+import javax.servlet.ServletResponse;
+import javax.servlet.http.HttpServletResponse;
+import java.io.IOException;
+
+/**
+ * A plugin interface that allows you to listen (and possibly mutate) the
+ * client requests to the Pulsar brokers.
+ *
+ * <p>Exceptions thrown by BrokerEventListener methods will be caught, logged, but
+ * not propagated further.
+ *
+ * <p>BrokerEventListener callbacks may be called from multiple threads. Interceptor
+ * implementation must ensure thread-safety, if needed.
+ */
+@Beta
+public interface BrokerEventListener extends AutoCloseable {
+
+    /**
+     * Called by the broker while new command incoming.
+     */
+    void onPulsarCommand(BaseCommand command, ServerCnx cnx);
+
+    /**
+     * Called by the web service while new request incoming.
+     */
+    void onWebServiceRequest(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException;
+
+    /**
+     * Initialize the broker event listener.
+     *
+     * @throws Exception when fail to initialize the broker event listener.
+     */
+    void initialize(ServiceConfiguration conf) throws Exception;
+
+    BrokerEventListener DISABLED = new BrokerEventListenerDisabled();
+
+    /**
+     * Broker event listener disabled implementation.
+     */
+    class BrokerEventListenerDisabled implements BrokerEventListener {
+
+        @Override
+        public void onPulsarCommand(BaseCommand command, ServerCnx cnx) {
+            //No-op
+        }
+
+        @Override
+        public void onWebServiceRequest(ServletRequest request, ServletResponse response, FilterChain chain) throws IOException, ServletException {
+            chain.doFilter(request, response);
+        }
+
+        @Override
+        public void initialize(ServiceConfiguration conf) throws Exception {
+            //No-op
+        }
+
+        @Override
+        public void close() {
+            //No-op
+        }
+    }
+
+    /**
+     * Close this broker event listener.
+     */
+    @Override
+    void close();
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/events/BrokerEventListenerDefinition.java

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.events;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+/**
+ * Metadata information about a broker event listener.
+ */
+@Data
+@NoArgsConstructor
+public class BrokerEventListenerDefinition {
+
+    /**
+     * The name of the broker event listener.
+     */
+    private String name;
+
+    /**
+     * The description of the broker event listener to be used for user help.
+     */
+    private String description;
+
+    /**
+     * The class name for the broker event listener.
+     */
+    private String listenerClass;
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/events/BrokerEventListenerDefinitions.java

@@ -0,0 +1,34 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.events;
+
+import lombok.Data;
+import lombok.experimental.Accessors;
+import java.util.Map;
+import java.util.TreeMap;
+
+/**
+ * The collection of broker event listener.
+ */
+@Data
+@Accessors(fluent = true)
+public class BrokerEventListenerDefinitions {
+
+    private final Map<String, BrokerEventListenerMetadata> listeners = new TreeMap<>();
+}

pulsar-broker/src/main/java/org/apache/pulsar/broker/events/BrokerEventListenerMetadata.java

@@ -0,0 +1,42 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.broker.events;
+
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.nio.file.Path;
+
+/**
+ * The metadata of broker event listener
+ */
+@Data
+@NoArgsConstructor
+public class BrokerEventListenerMetadata {
+
+    /**
+     * The definition of the broker event listener.
+     */
+    private BrokerEventListenerDefinition definition;
+
+    /**
+     * The path to the handler package.
+     */
+    private Path archivePath;
+}