[improve][broker] Improve the extensibility of the TopicBundleAssignmentStrategy interface class

[rayluoluo]

Search before asking

• I searched in the issues and found nothing similar.

Motivation

During the lookup process of the pulsar broker, the hash value needs to be calculated based on the topic name. For example:

1. Bundle obtaining triggered by a lookup request:

   Lookup -> NamespaceService#getBrokerServiceUrlAsync -> NamespaceService#getBundleAsync ->
   NamespaceBundles#findBundle -> TopicBundleAssignmentStrategy#findBundle -> NamespaceBundles#getBundle(long hash)

2. When loading a topic, the broker needs to determine whether it owns the topic partition.

   PulsarService#loadNamespaceTopics -> NamespaceBundle#includes -> NamespaceBundleFactory#getLongHashCode -> NamespaceBundle.keyRange#contains(long hash)

The current code implementation has the following problems:

1. The hash algorithm is fixed. When the load balancing algorithm is extended, the bundle to which the partition belongs cannot be adjusted. As a result, other algorithms such as RoundRobin cannot be extended.

   The NamespaceBundleFactory#getLongHashCode method uses a fixed algorithm to calculate the hash value. Therefore, it is difficult to extend the implementation of the TopicBundleAssignmentStrategy interface class that uses different hash algorithms without modifying the NamespaceBundleFactory#getLongHashCode method, which violates the open and closed principles.

2. Bad code smell (shot-like modification): The hash algorithm is implemented in the findBundle and getLongHashCode methods. The system must ensure that the calculated hash results are the same. Otherwise, split-brain occurs in the cluster. Therefore, if the hash algorithm needs to be modified, the code has a bad smell.

   • Lookup request:

     Take the default implementation class ConsistentHashingTopicBundleAssigner of the TopicBundleAssignmentStrategy interface class as an example. During the lookup process, the hash value is calculated in ConsistentHashingTopicBundleAssigner#findBundle.

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

     Lines 25 to 40 in 1967a93

     	public class ConsistentHashingTopicBundleAssigner implements TopicBundleAssignmentStrategy {
     	@Override
     	public NamespaceBundle findBundle(TopicName topicName, NamespaceBundles namespaceBundles) {
     	long hashCode = Hashing.crc32().hashString(topicName.toString(), StandardCharsets.UTF_8).padToLong();
     	NamespaceBundle bundle = namespaceBundles.getBundle(hashCode);
     	if (topicName.getDomain().equals(TopicDomain.non_persistent)) {
     	bundle.setHasNonPersistentTopic(true);
     	}
     	return bundle;
     	}
     	@Override
     	public void init(PulsarService pulsarService) {
     	}
     	}

   • When a topic is loaded, the hash value is calculated in the NamespaceBundleFactory#getLongHashCode method to determine whether the current broker owns the topic.

     pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

     Line 204 in 1967a93

     this.bundleFactory = new NamespaceBundleFactory(pulsar, Hashing.crc32());

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

     Lines 78 to 79 in 1967a93

     	public NamespaceBundleFactory(PulsarService pulsar, HashFunction hashFunc) {
     	this.hashFunc = hashFunc;

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

     Lines 294 to 296 in 1967a93

     	public long getLongHashCode(String name) {
     	return this.hashFunc.hashString(name, StandardCharsets.UTF_8).padToLong();
     	}

Solution

It is recommended that the implementation of the NamespaceBundleFactory#getLongHashCode method be moved to the implementation class of the interface TopicBundleAssignmentStrategy. Therefore, we may add a new method long getHashCode(String name) to the TopicBundleAssignmentStrategy interface class. The implementation of the hash algorithm is no longer fixed in the NamespaceBundleFactory#getLongHashCode method. Instead, the getHashCode method implemented by different algorithms is invoked.

Alternatives

No response

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!