Make dispatcher pluggable

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

@@ -626,6 +626,62 @@ public class ServiceConfiguration implements PulsarConfiguration {
             + " non-backlog consumers as well.")
     private boolean dispatchThrottlingOnNonBacklogConsumerEnabled = false;
 
+    // <-- Pluggable dispatcher -->
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Path to a nar file containing all customized dispatchers if we want to use customized dispatcher."
+    )
+    private String dispatcherNarPath = null;
+
+    @FieldContext(
+        category = CATEGORY_SERVER,
+        doc = "Customized dispatcher to use for persistent Exclusive subscription mode."
+    )
+    private String persistentDispatcherExclusive = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for persistent Shared subscription mode."
+    )
+    private String persistentDispatcherShared = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for persistent Failover subscription mode."
+    )
+    private String persistentDispatcherFailover = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for persistent Key_Shared subscription mode."
+    )
+    private String persistentDispatcherKeyShared = null;
+
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for non-persistent Exclusive subscription mode."
+    )
+    private String nonpersistentDispatcherExclusive = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for non-persistent Shared subscription mode."
+    )
+    private String nonpersistentDispatcherShared = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for non-persistent Failover subscription mode."
+    )
+    private String nonpersistentDispatcherFailover = null;
+
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            doc = "Customized dispatcher to use for non-persistent Key_Shared subscription mode."
+    )
+    private String nonpersistentDispatcherKeyShared = null;
+
     // <-- dispatcher read settings -->
     @FieldContext(
         dynamic = true,

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

@@ -104,6 +104,8 @@
 import org.apache.pulsar.broker.service.BrokerServiceException.PersistenceException;
 import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
 import org.apache.pulsar.broker.service.BrokerServiceException.ServiceUnitNotReadyException;
+import org.apache.pulsar.broker.service.dispatcher.Dispatcher;
+import org.apache.pulsar.broker.service.dispatcher.DispatcherUtils;
 import org.apache.pulsar.broker.service.nonpersistent.NonPersistentTopic;
 import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
 import org.apache.pulsar.broker.service.persistent.PersistentDispatcherMultipleConsumers;
@@ -446,6 +448,8 @@ public void start() throws Exception {
         // register listener to capture zk-latency
         ClientCnxnAspect.addListener(zkStatsListener);
         ClientCnxnAspect.registerExecutor(pulsar.getExecutor());
+
+        DispatcherUtils.init(serviceConfig.getDispatcherNarPath());
     }
 
     protected void startStatsUpdater(int statsUpdateInitailDelayInSecs, int statsUpdateFrequencyInSecs) {
@@ -680,6 +684,7 @@ public void close() throws IOException {
         ClientCnxnAspect.registerExecutor(null);
         topicOrderedExecutor.shutdown();
         delayedDeliveryTrackerFactory.close();
+        DispatcherUtils.close();
         if (topicPublishRateLimiterMonitor != null) {
             topicPublishRateLimiterMonitor.shutdown();
         }

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

@@ -189,7 +189,7 @@ public String consumerName() {
         return consumerName;
     }
 
-    void notifyActiveConsumerChange(Consumer activeConsumer) {
+    public void notifyActiveConsumerChange(Consumer activeConsumer) {
         if (log.isDebugEnabled()) {
             log.debug("notify consumer {} - that [{}] for subscription {} has new active consumer : {}",
                 consumerId, topicName, subscription.getName(), activeConsumer);

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

@@ -24,6 +24,7 @@
 import org.apache.bookkeeper.mledger.Entry;
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.dispatcher.Dispatcher;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 import org.apache.pulsar.common.api.proto.PulsarMarkers.ReplicatedSubscriptionsSnapshot;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/service/dispatcher/AbstractBaseDispatcher.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.apache.pulsar.broker.service;
+package org.apache.pulsar.broker.service.dispatcher;
 
 import io.netty.buffer.ByteBuf;
 import java.io.IOException;
@@ -30,6 +30,11 @@
 import org.apache.bookkeeper.mledger.Position;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
 import org.apache.commons.lang3.tuple.Pair;
+import org.apache.pulsar.broker.service.EntryBatchIndexesAcks;
+import org.apache.pulsar.broker.service.EntryBatchSizes;
+import org.apache.pulsar.broker.service.SendMessageInfo;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.dispatcher.Dispatcher;
 import org.apache.pulsar.broker.service.persistent.PersistentTopic;
 import org.apache.pulsar.common.api.proto.PulsarApi;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandAck.AckType;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherMultipleConsumers.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/service/dispatcher/AbstractDispatcherMultipleConsumers.java

@@ -16,13 +16,16 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.broker.service;
+package org.apache.pulsar.broker.service.dispatcher;
 
 import com.carrotsearch.hppc.ObjectHashSet;
 import com.carrotsearch.hppc.ObjectSet;
 import java.util.Random;
 import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
+
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.Subscription;
 import org.apache.pulsar.broker.service.persistent.PersistentStickyKeyDispatcherMultipleConsumers;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 import org.slf4j.Logger;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/service/dispatcher/AbstractDispatcherSingleActiveConsumer.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.broker.service;
+package org.apache.pulsar.broker.service.dispatcher;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import java.util.List;
@@ -26,8 +26,14 @@
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicIntegerFieldUpdater;
 import java.util.concurrent.atomic.AtomicReferenceFieldUpdater;
+
+import org.apache.pulsar.broker.service.BrokerServiceException;
 import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerBusyException;
 import org.apache.pulsar.broker.service.BrokerServiceException.ServerMetadataException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.HashRangeExclusiveStickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.StickyKeyConsumerSelector;
+import org.apache.pulsar.broker.service.Subscription;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Dispatcher.java → pulsar-broker/src/main/java/org/apache/pulsar/broker/service/dispatcher/Dispatcher.java

@@ -16,12 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.pulsar.broker.service;
+package org.apache.pulsar.broker.service.dispatcher;
 
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
 import org.apache.bookkeeper.mledger.impl.PositionImpl;
+import org.apache.pulsar.broker.service.BrokerServiceException;
+import org.apache.pulsar.broker.service.Consumer;
+import org.apache.pulsar.broker.service.RedeliveryTracker;
 import org.apache.pulsar.broker.service.persistent.DispatchRateLimiter;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 import org.apache.pulsar.common.api.proto.PulsarApi.MessageMetadata;
@@ -119,4 +122,8 @@ default void markDeletePositionMoveForward() {
         // No-op
     }
 
+    default void init(DispatcherConfiguration dispatcherConfiguration) {
+        // No-op
+    }
+
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/dispatcher/DispatcherConfiguration.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.service.dispatcher;
+
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import org.apache.bookkeeper.mledger.ManagedCursor;
+import org.apache.pulsar.broker.service.Subscription;
+import org.apache.pulsar.broker.service.Topic;
+import org.apache.pulsar.common.api.proto.PulsarApi;
+
+/**
+ * Holds configuration for creating various kinds of dispatchers.
+ */
+@Data
+@AllArgsConstructor
+public class DispatcherConfiguration {
+    private PulsarApi.CommandSubscribe.SubType subType;
+
+    private ManagedCursor cursor;
+
+    private int partitionIndex;
+
+    private Topic topic;
+
+    private Subscription subscription;
+
+    private PulsarApi.KeySharedMeta ksm;
+}