[improve][broker] PIP-392: Add configuration to enable consistent hashing to select active consumer for partitioned topic

conf/broker.conf

@@ -243,6 +243,10 @@ messageExpiryCheckIntervalInMinutes=5
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. 
+# For non-partitioned topics, consistent hashing is used by default.
+activeConsumerFailoverConsistentHashing=false
+
 # How long to delete inactive subscriptions from last consuming
 # When it is 0, inactive subscriptions are not deleted automatically
 subscriptionExpirationTimeMinutes=0

conf/standalone.conf

@@ -152,6 +152,10 @@ maxMessageSizeCheckIntervalInSeconds=60
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. 
+# For non-partitioned topics, consistent hashing is used by default.
+activeConsumerFailoverConsistentHashing=false
+
 # How long to delete inactive subscriptions from last consuming
 # When it is 0, inactive subscriptions are not deleted automatically
 subscriptionExpirationTimeMinutes=0

deployment/terraform-ansible/templates/broker.conf

@@ -148,6 +148,10 @@ messageExpiryCheckIntervalInMinutes=5
 # How long to delay rewinding cursor and dispatching messages when active consumer is changed
 activeConsumerFailoverDelayTimeMillis=1000
 
+# Enable consistent hashing for selecting the active consumer in partitioned topics with Failover subscription type. 
+# For non-partitioned topics, consistent hashing is used by default.
+activeConsumerFailoverConsistentHashing=false
+
 # How long to delete inactive subscriptions from last consuming
 # When it is 0, inactive subscriptions are not deleted automatically
 subscriptionExpirationTimeMinutes=0

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

@@ -760,6 +760,13 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
         doc = "How long to delay rewinding cursor and dispatching messages when active consumer is changed"
     )
     private int activeConsumerFailoverDelayTimeMillis = 1000;
+    @FieldContext(
+            category = CATEGORY_POLICIES,
+            doc = "Enable consistent hashing for selecting the active consumer in partitioned "
+                    + "topics with Failover subscription type."
+                    + "For non-partitioned topics, consistent hashing is used by default."
+    )
+    private boolean activeConsumerFailoverConsistentHashing = false;
     @FieldContext(
             category = CATEGORY_POLICIES,
             doc = "Maximum time to spend while scanning a subscription to calculate the accurate backlog"

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

@@ -126,7 +126,7 @@ protected boolean pickAndScheduleActiveConsumer() {
                 }
             }
         }
-        int index = partitionIndex >= 0
+        int index = partitionIndex >= 0 && !serviceConfig.isActiveConsumerFailoverConsistentHashing()
                 ? partitionIndex % consumersSize
                 : peekConsumerIndexFromHashRing(makeHashRing(consumersSize));
 

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/TopicsConsumerImplTest.java

@@ -1380,6 +1380,39 @@ public void testTopicsDistribution() throws Exception {
         }
     }
 
+    @Test
+    public void testPartitionedTopicDistribution() throws Exception {
+        this.conf.setActiveConsumerFailoverConsistentHashing(true);
+        final String topic = "partitioned-topics-distribution";
+        final int topicCount = 100;
+        final int consumers = 10;
+
+        for (int i = 0; i < topicCount; i++) {
+            admin.topics().createPartitionedTopic(topic + "-" + i, 1);
+        }
+
+        CustomizedConsumerEventListener eventListener = new CustomizedConsumerEventListener();
+
+        List<Consumer<?>> consumerList = new ArrayList<>(consumers);
+        for (int i = 0; i < consumers; i++) {
+            consumerList.add(pulsarClient.newConsumer()
+                    .topics(IntStream.range(0, topicCount).mapToObj(j -> topic + "-" + j).toList())
+                    .subscriptionType(SubscriptionType.Failover)
+                    .subscriptionName("my-sub")
+                    .consumerName("consumer-" + i)
+                    .consumerEventListener(eventListener)
+                    .subscribe());
+        }
+
+        log.info("Topics are distributed to consumers as {}", eventListener.getActiveConsumers());
+        Map<String, Integer> assigned = new HashMap<>();
+        eventListener.getActiveConsumers().forEach((k, v) -> assigned.compute(v, (t, c) -> c == null ? 1 : ++ c));
+        assertEquals(assigned.size(), consumers);
+        for (Consumer<?> consumer : consumerList) {
+            consumer.close();
+        }
+    }
+
     private static class CustomizedConsumerEventListener implements ConsumerEventListener {
 
         private final Map<String, String> activeConsumers = new HashMap<>();