[feat][broker]PIP-255 Part-2: Make the partition assignment strategy pluggable

conf/broker.conf

@@ -1308,6 +1308,9 @@ loadBalancerOverrideBrokerNicSpeedGbps=
 # Name of load manager to use
 loadManagerClassName=org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl
 
+#Name of topic bundle assignment strategy to use
+topicBundleAssignmentStrategy=org.apache.pulsar.common.naming.ConsistentHashingTopicBundleAssigner
+
 # Supported algorithms name for namespace bundle split.
 # "range_equally_divide" divides the bundle into two parts with the same hash range size.
 # "topic_count_equally_divide" divides the bundle into two parts with the same topics count.

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

@@ -2477,6 +2477,11 @@ The delayed message index time step(in seconds) in per bucket snapshot segment,
         doc = "Name of load manager to use"
     )
     private String loadManagerClassName = "org.apache.pulsar.broker.loadbalance.impl.ModularLoadManagerImpl";
+
+
+    @FieldContext(category = CATEGORY_LOAD_BALANCER, doc = "Name of topic bundle assignment strategy to use")
+    private String topicBundleAssignmentStrategy =
+            "org.apache.pulsar.common.naming.ConsistentHashingTopicBundleAssigner";
     @FieldContext(
         dynamic = true,
         category = CATEGORY_LOAD_BALANCER,

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/ConsistentHashingTopicBundleAssigner.java

@@ -0,0 +1,43 @@
+/*
+ * 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.common.naming;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+
+public class ConsistentHashingTopicBundleAssigner implements TopicBundleAssignmentStrategy {
+    private NamespaceService namespaceService;
+    @Override
+    public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+        long hashCode = namespaceService.getNamespaceBundleFactory().getLongHashCode(topicName.toString());
+        NamespaceBundle bundle = namespaceBundles.getBundle(hashCode);
+        checkArgument(bundle.getNamespaceObject().equals(topicName.getNamespaceObject()));
+        if (topicName.getDomain().equals(TopicDomain.non_persistent)) {
+            bundle.setHasNonPersistentTopic(true);
+        }
+        return bundle;
+    }
+
+    @Override
+    public void init(NamespaceService namespaceService, PulsarAdmin pulsarAdmin, ServiceConfiguration configuration) {
+        this.namespaceService = namespaceService;
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundleFactory.java

@@ -377,6 +377,10 @@ public boolean canSplitBundle(NamespaceBundle bundle) {
         return range.upperEndpoint() - range.lowerEndpoint() > 1;
     }
 
+    public PulsarService getPulsar() {
+        return pulsar;
+    }
+
     public static void validateFullRange(SortedSet<String> partitions) {
         checkArgument(partitions.first().equals(FIRST_BOUNDARY) && partitions.last().equals(LAST_BOUNDARY));
     }

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/NamespaceBundles.java

@@ -46,7 +46,7 @@ public class NamespaceBundles {
     public static final Long FULL_LOWER_BOUND = 0x00000000L;
     public static final Long FULL_UPPER_BOUND = 0xffffffffL;
     private final NamespaceBundle fullBundle;
-
+    private final TopicBundleAssignmentStrategy topicBundleAssignmentStrategy;
     private final Optional<Pair<LocalPolicies, Long>> localPolicies;
 
     public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
@@ -66,7 +66,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
         this.factory = Objects.requireNonNull(factory);
         this.localPolicies = localPolicies;
         checkArgument(partitions.length > 0, "Can't create bundles w/o partition boundaries");
-
+        this.topicBundleAssignmentStrategy = TopicBundleAssignmentFactory.create(factory.getPulsar());
         // calculate bundles based on partition boundaries
         this.bundles = new ArrayList<>();
         fullBundle = new NamespaceBundle(nsname,
@@ -94,13 +94,7 @@ public NamespaceBundles(NamespaceName nsname, NamespaceBundleFactory factory,
     }
 
     public NamespaceBundle findBundle(TopicName topicName) {
-        checkArgument(this.nsname.equals(topicName.getNamespaceObject()));
-        long hashCode = factory.getLongHashCode(topicName.toString());
-        NamespaceBundle bundle = getBundle(hashCode);
-        if (topicName.getDomain().equals(TopicDomain.non_persistent)) {
-            bundle.setHasNonPersistentTopic(true);
-        }
-        return bundle;
+        return topicBundleAssignmentStrategy.findBundle(topicName, this);
     }
 
     public List<NamespaceBundle> getBundles() {

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentFactory.java

@@ -0,0 +1,41 @@
+/*
+ * 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.common.naming;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.common.util.Reflections;
+
+public class TopicBundleAssignmentFactory {
+
+    public static TopicBundleAssignmentStrategy create(PulsarService pulsar) {
+        ServiceConfiguration conf = pulsar.getConfiguration();
+        NamespaceService namespaceService = pulsar.getNamespaceService();
+        try {
+            TopicBundleAssignmentStrategy strategy = Reflections.createInstance(conf.getTopicBundleAssignmentStrategy(),
+                    TopicBundleAssignmentStrategy.class, Thread.currentThread().getContextClassLoader());
+            strategy.init(namespaceService, pulsar.getAdminClient(), pulsar.getConfiguration());
+            return strategy;
+        } catch (Exception e) {
+            throw new RuntimeException(
+                    "Could not load TopicBundleAssignmentStrategy:" + conf.getTopicBundleAssignmentStrategy(), e);
+        }
+    }
+}

pulsar-broker/src/main/java/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategy.java

@@ -0,0 +1,29 @@
+/*
+ * 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.common.naming;
+
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+
+public interface TopicBundleAssignmentStrategy {
+    NamespaceBundle findBundle(TopicName topicName,  NamespaceBundles namespaceBundles);
+
+    void init(NamespaceService namespaceService, PulsarAdmin pulsarAdmin, ServiceConfiguration configuration);
+}

pulsar-broker/src/test/org/apache/pulsar/common/naming/TopicBundleAssignmentStrategyTest.java

@@ -0,0 +1,65 @@
+/*
+ * 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.common.naming;
+
+import static org.mockito.Mockito.doReturn;
+import static org.mockito.Mockito.mock;
+
+import com.google.common.collect.BoundType;
+import com.google.common.collect.Range;
+
+import org.apache.pulsar.broker.PulsarService;
+import org.apache.pulsar.broker.ServiceConfiguration;
+import org.apache.pulsar.broker.namespace.NamespaceService;
+import org.apache.pulsar.client.admin.PulsarAdmin;
+import org.testng.Assert;
+import org.testng.annotations.Test;
+
+@Test(groups = "broker-naming")
+public class TopicBundleAssignmentStrategyTest {
+    @Test
+    public void testStrategyFactory() {
+        ServiceConfiguration conf = new ServiceConfiguration();
+        conf.setTopicBundleAssignmentStrategy(
+                "org.apache.pulsar.common.naming.TopicBundleAssignmentStrategyTest$TestStrategy");
+        PulsarService pulsarService = mock(PulsarService.class);
+        doReturn(conf).when(pulsarService).getConfiguration();
+        TopicBundleAssignmentStrategy strategy = TopicBundleAssignmentFactory.create(pulsarService);
+        NamespaceBundle bundle = strategy.findBundle(null, null);
+        Range<Long> keyRange = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+        String range = String.format("0x%08x_0x%08x", keyRange.lowerEndpoint(), keyRange.upperEndpoint());
+        Assert.assertEquals(bundle.getBundleRange(), range);
+        Assert.assertEquals(bundle.getNamespaceObject(), NamespaceName.get("my/test"));
+    }
+
+    public static class TestStrategy implements TopicBundleAssignmentStrategy {
+        @Override
+        public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
+            Range<Long> range = Range.range(0L, BoundType.CLOSED, 0xffffffffL, BoundType.CLOSED);
+            return new NamespaceBundle(NamespaceName.get("my/test"), range,
+                    mock(NamespaceBundleFactory.class));
+        }
+
+        @Override
+        public void init(NamespaceService namespaceService, PulsarAdmin pulsarAdmin,
+                         ServiceConfiguration configuration) {
+        }
+    }
+
+}