You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/04/22 02:10:14 UTC
[GitHub] [pulsar] merlimat opened a new pull request #6791: Use consistent hashing in KeyShared distribution
merlimat opened a new pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791
### Motivation
The current implementation of KeyShared subscriptions uses a mechanism to divide they hash space across the available consumers. This is based on dividing the currently assigned hash ranges when a new consumer joins or leaves.
There are few problems with the current approach:
1. When adding a new consumer, the bigger range is split to make space for the new consumer. That means that when adding 3 consumers, 1 of them will "own" a hash range that is double in size compared to the other 2 consumers and consequently it will receive twice the traffic. This is not terrible, but not ideal either.
2. When removing consumers, the range for the removed consumer will always be assigned to the next consumer. The new hash distribution really depends on the sequence upon which the consumers are removed. If one is unlucky, the traffic will be very heavily skewed having situations where 1 consumer is getting >90% of the traffic.
This is an example of removing consumers in sequence, with attached the size of their respective hash ranges:
```
Removed consumer from rangeMap: {c1=8192, c10=4096, c3=4096, c4=8192, c5=4096, c6=8192, c7=16384, c8=8192, c9=4096}
Removed consumer from rangeMap: {c1=8192, c10=4096, c4=8192, c5=4096, c6=12288, c7=16384, c8=8192, c9=4096}
Removed consumer from rangeMap: {c1=8192, c10=4096, c5=4096, c6=12288, c7=16384, c8=16384, c9=4096}
Removed consumer from rangeMap: {c1=8192, c10=8192, c6=12288, c7=16384, c8=16384, c9=4096}
Removed consumer from rangeMap: {c1=24576, c10=8192, c6=12288, c7=16384, c9=4096}
Removed consumer from rangeMap: {c1=24576, c10=8192, c7=28672, c9=4096}
Removed consumer from rangeMap: {c1=53248, c10=8192, c9=4096}
```
As you can see, `c1` will take most of the traffic.
Most likely it will not be able to process all the messages and the backlog builds up.
### Modifications
* No functional difference from user perspective
* Use consistent hashing mechanism to assign keys to consumers. This will ensure even distribution without the degradation in the corner cases.
* Number of points in the ring is configurable, default=100.
* Refactored current unit test. The test are currently duplicating the logic of the implementation and checking the a messages is placed on the bucket for one consumer. Of course it works, since it's the same code executed on both sides. But, instead, the test should focus on the contract of the feature: message should arrive in order, there should be "decent" sharing of load across consumers.
* @codelipenghui I've removed the `selectByIndex()`. In my opinion there's absolutely no difference in efficiency/performance as I've also explained on https://github.com/apache/pulsar/pull/6647#issuecomment-617497271. I'm happy to discuss more about it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413398739
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeAutoSplitStickyKeyConsumerSelector.java
##########
@@ -18,86 +18,64 @@
*/
package org.apache.pulsar.broker.service;
-import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
/**
* This is a consumer selector based fixed hash range.
*
- * 1.Each consumer serves a fixed range of hash value
- * 2.The whole range of hash value could be covered by all the consumers.
- * 3.Once a consumer is removed, the left consumers could still serve the whole range.
- *
- * Initializing with a fixed hash range, by default 2 << 5.
- * First consumer added, hash range looks like:
- *
- * 0 -> 65536(consumer-1)
- *
- * Second consumer added, will find a biggest range to split:
- *
- * 0 -> 32768(consumer-2) -> 65536(consumer-1)
- *
- * While a consumer removed, The range for this consumer will be taken over
- * by other consumer, consumer-2 removed:
- *
- * 0 -> 65536(consumer-1)
- *
- * In this approach use skip list map to maintain the hash range and consumers.
- *
- * Select consumer will return the ceiling key of message key hashcode % range size.
- *
+ * The implementation uses consistent hashing to evenly split, the
+ * number of keys assigned to each consumer.
*/
public class HashRangeAutoSplitStickyKeyConsumerSelector implements StickyKeyConsumerSelector {
- private final int rangeSize;
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
- private final ConcurrentSkipListMap<Integer, Consumer> rangeMap;
- private final Map<Consumer, Integer> consumerRange;
+ // Consistent-Hash ring
+ private final NavigableMap<Integer, Consumer> hashRing;
- public HashRangeAutoSplitStickyKeyConsumerSelector() {
- this(DEFAULT_RANGE_SIZE);
- }
+ private final int numberOfPOints;
- public HashRangeAutoSplitStickyKeyConsumerSelector(int rangeSize) {
- if (rangeSize < 2) {
- throw new IllegalArgumentException("range size must greater than 2");
- }
- if (!is2Power(rangeSize)) {
- throw new IllegalArgumentException("range size must be nth power with 2");
- }
- this.rangeMap = new ConcurrentSkipListMap<>();
- this.consumerRange = new HashMap<>();
- this.rangeSize = rangeSize;
+ public HashRangeAutoSplitStickyKeyConsumerSelector(int numberOfPOints) {
+ this.hashRing = new TreeMap<>();
+ this.numberOfPOints = numberOfPOints;
}
@Override
- public synchronized void addConsumer(Consumer consumer) throws ConsumerAssignException {
- if (rangeMap.size() == 0) {
- rangeMap.put(rangeSize, consumer);
- consumerRange.put(consumer, rangeSize);
- } else {
- splitRange(findBiggestRange(), consumer);
+ public void addConsumer(Consumer consumer) throws ConsumerAssignException {
+ rwLock.writeLock().lock();
+ try {
+ // Insert multiple points on the hash ring for every consumer
+ // The points are deterministically added based on the hash of the consumer name
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.put(hash, consumer);
+ }
+ } finally {
+ rwLock.writeLock().unlock();
}
}
@Override
- public synchronized void removeConsumer(Consumer consumer) {
- Integer removeRange = consumerRange.remove(consumer);
- if (removeRange != null) {
- if (removeRange == rangeSize && rangeMap.size() > 1) {
- Map.Entry<Integer, Consumer> lowerEntry = rangeMap.lowerEntry(removeRange);
- rangeMap.put(removeRange, lowerEntry.getValue());
- rangeMap.remove(lowerEntry.getKey());
- consumerRange.put(lowerEntry.getValue(), removeRange);
- } else {
- rangeMap.remove(removeRange);
+ public void removeConsumer(Consumer consumer) {
+ rwLock.writeLock().lock();
+ try {
+ // Remove all the points that were added for this consumer
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.remove(hash, consumer);
Review comment:
It's necessary because there can be a hash conflict. When that happens, the point will only be assigned to one consumer.
It is not a problem since we have multiple points in the ring, though when we remove `c1` we need to be sure we're not removing a point that was actually assigned to `c2`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618223059
> Since the current algorithm has been used in many production environments already, we should be concerned about removing an existing implementation. Providing a way to opt-in a different algorithm should be the right approach to go - just like what we did for other algorithms like bundle split, load shedder and etc.
The current implementation is severely broken. This PR is fixing the bug. I don't see an easy way to fix it based on the splitting range design.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413395480
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeAutoSplitStickyKeyConsumerSelector.java
##########
@@ -18,86 +18,64 @@
*/
package org.apache.pulsar.broker.service;
-import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
/**
* This is a consumer selector based fixed hash range.
*
- * 1.Each consumer serves a fixed range of hash value
- * 2.The whole range of hash value could be covered by all the consumers.
- * 3.Once a consumer is removed, the left consumers could still serve the whole range.
- *
- * Initializing with a fixed hash range, by default 2 << 5.
- * First consumer added, hash range looks like:
- *
- * 0 -> 65536(consumer-1)
- *
- * Second consumer added, will find a biggest range to split:
- *
- * 0 -> 32768(consumer-2) -> 65536(consumer-1)
- *
- * While a consumer removed, The range for this consumer will be taken over
- * by other consumer, consumer-2 removed:
- *
- * 0 -> 65536(consumer-1)
- *
- * In this approach use skip list map to maintain the hash range and consumers.
- *
- * Select consumer will return the ceiling key of message key hashcode % range size.
- *
+ * The implementation uses consistent hashing to evenly split, the
+ * number of keys assigned to each consumer.
*/
public class HashRangeAutoSplitStickyKeyConsumerSelector implements StickyKeyConsumerSelector {
- private final int rangeSize;
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
- private final ConcurrentSkipListMap<Integer, Consumer> rangeMap;
- private final Map<Consumer, Integer> consumerRange;
+ // Consistent-Hash ring
+ private final NavigableMap<Integer, Consumer> hashRing;
- public HashRangeAutoSplitStickyKeyConsumerSelector() {
- this(DEFAULT_RANGE_SIZE);
- }
+ private final int numberOfPOints;
- public HashRangeAutoSplitStickyKeyConsumerSelector(int rangeSize) {
- if (rangeSize < 2) {
- throw new IllegalArgumentException("range size must greater than 2");
- }
- if (!is2Power(rangeSize)) {
- throw new IllegalArgumentException("range size must be nth power with 2");
- }
- this.rangeMap = new ConcurrentSkipListMap<>();
- this.consumerRange = new HashMap<>();
- this.rangeSize = rangeSize;
+ public HashRangeAutoSplitStickyKeyConsumerSelector(int numberOfPOints) {
+ this.hashRing = new TreeMap<>();
+ this.numberOfPOints = numberOfPOints;
}
@Override
- public synchronized void addConsumer(Consumer consumer) throws ConsumerAssignException {
- if (rangeMap.size() == 0) {
- rangeMap.put(rangeSize, consumer);
- consumerRange.put(consumer, rangeSize);
- } else {
- splitRange(findBiggestRange(), consumer);
+ public void addConsumer(Consumer consumer) throws ConsumerAssignException {
+ rwLock.writeLock().lock();
+ try {
+ // Insert multiple points on the hash ring for every consumer
+ // The points are deterministically added based on the hash of the consumer name
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.put(hash, consumer);
+ }
+ } finally {
+ rwLock.writeLock().unlock();
}
}
@Override
- public synchronized void removeConsumer(Consumer consumer) {
- Integer removeRange = consumerRange.remove(consumer);
- if (removeRange != null) {
- if (removeRange == rangeSize && rangeMap.size() > 1) {
- Map.Entry<Integer, Consumer> lowerEntry = rangeMap.lowerEntry(removeRange);
- rangeMap.put(removeRange, lowerEntry.getValue());
- rangeMap.remove(lowerEntry.getKey());
- consumerRange.put(lowerEntry.getValue(), removeRange);
- } else {
- rangeMap.remove(removeRange);
+ public void removeConsumer(Consumer consumer) {
+ rwLock.writeLock().lock();
+ try {
+ // Remove all the points that were added for this consumer
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.remove(hash, consumer);
Review comment:
When removing consumers from the hash ring, is it necessary to check if the value of the key we are removing is also equals the consumer? In what case, will that not be true?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
sijie commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618267027
> The current implementation is severely broken. This PR is fixing the bug. I don't see an easy way to fix it based on the splitting range design.
I see there is a problem in the current algorithm when scaling down the number of consumers. But I don't think its broken. In most of the production, people usually don't scale down the number of consumers. They usually scale up the number of consumers. So the extreme case you described in the description probably wasn't really noticed. This is the caveat of the current algorithm. We can find ways to improve it or other people in the community can find a way to improve it if you don't know how to fix it.
If you have a better algorithm to resolve the key distribution and the screw problem during split-and-merge, that's great and the community is excited about it. Then let's introduce that algorithm. This allows people to slowly migrate to a new algorithm and fall back as needed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-630836016
@merlimat I'm planning for 2.6.0 release and It's great if we can onboard the consistent hashing distribution in KeyShared subscription.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-635104421
/pulsarbot run-failure-checks
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413405395
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentKeySharedSubscriptionTest.java
##########
@@ -1,482 +0,0 @@
-/**
Review comment:
Why are we deleting this whole test?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie edited a comment on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
sijie edited a comment on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618267027
> The current implementation is severely broken. This PR is fixing the bug. I don't see an easy way to fix it based on the splitting range design.
I see there is a problem in the current algorithm when scaling down the number of consumers. But I don't think its broken. In most of the production, people usually don't scale down the number of consumers. They usually scale up the number of consumers. So the extreme case you described in the description probably wasn't really noticed. This is the caveat of the current algorithm. We can find ways to improve it or other people in the community can find a way to improve it.
If you have a better algorithm to resolve the key distribution and the screw problem during split-and-merge, that's great and the community is excited about it. Then let's introduce that algorithm. This allows people to slowly migrate to a new algorithm and fall back as needed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618531886
> I see there is a problem in the current algorithm when scaling down the number of consumers. But I don't think its broken.
The Key Shared subscriptions has 2 stated goals:
1. Distribute the load evenly across available consumers
2. Retain ordering
The current implementation cannot guarantee either of the 2:
1. Apart from the scale down I've seen 1 consumer out of 10 being active after a rolling restart
2. A simple sequence of "add new consumer" and "remove a different consumer" is breaking the ordering guarantee.
> So the extreme case you described in the description probably wasn't really noticed.
The first user I saw using this feature for a very simple use case kept going into this state after each rolling restart.
For sure, bugs are always around the corner, though once identified we should should strive to fix them.
> If you have a better algorithm to resolve the key distribution and the screw problem during split-and-merge,
That's what I said: There is no easy way to do it in that mode and that's the reason why the textbook solution for this kind of problems is to use consistent hashing. This is not something I came up with.
So, if you're asking "do you have a better algorithm to implement the goals of KeyShared subscription", the answer is: yes, consistent hashing.
Instead, if the question is "do you have a better algorithm to resolve the key distribution and the screw problem during split-and-merge?", my answer is no and: why is that a requirement? We just need one solution that fulfills the 2 goals (traffic distribution and ordering).
Going back to the original question on having a configuration switch. We can do that if there are concerns on this implementation, though I think that, at least by default, the behavior should not have "known" blind spots.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-636409528
ping @sijie Please help take a look at this PR again.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618109262
@codelipenghui AUTO_SPLIT and CONSISTENT_HASHING are essentially 2 different implementations for the same stated behavior. In my view, we should just give the user one good default option. That option should work across all corner cases.
Do you see any particular downside in using consistent hashing compared to the autosplit approach?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-617958166
> I need to rethink how to leverage this idea on the consistent hashing approach. If you have any idea, please ping me, thanks.
@codelipenghui Thanks for pointing to this. I have a couple of rough ideas. Let me think it through it a bit more and I'll chime in on that issue.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-617632190
@merlimat I'm working on fixing ordering dispatching for consumers adds and removes. This is issue #6554 for it and we have discussed in the slack channel https://apache-pulsar.slack.com/archives/C5Z4T36F7/p1585579082105200
I want to introduce a concept named fenced range, the main idea is for do not block the whole dispatching task when consumers changed(especially add consumer).
The general approach is when a new consumer added, the split range becomes fenced. And every split range has a buffer(it's more like `messagesToRedeliver` in the dispatcher). The messages that dispatch to this range will add the buffer and do not dispatch to the consumer. When the owner consumer of the split range takes all messages of its receiver queue, the fenced range become un-fenced, so that we can redeliver these messages. Every fenced range has a fence timeout, this can be controlled by users.
I need to rethink how to leverage this idea on the consistent hashing approach. If you have any idea, please ping me, thanks.
The PR looks good.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413394891
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeAutoSplitStickyKeyConsumerSelector.java
##########
@@ -18,86 +18,64 @@
*/
package org.apache.pulsar.broker.service;
-import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
/**
* This is a consumer selector based fixed hash range.
*
- * 1.Each consumer serves a fixed range of hash value
- * 2.The whole range of hash value could be covered by all the consumers.
- * 3.Once a consumer is removed, the left consumers could still serve the whole range.
- *
- * Initializing with a fixed hash range, by default 2 << 5.
- * First consumer added, hash range looks like:
- *
- * 0 -> 65536(consumer-1)
- *
- * Second consumer added, will find a biggest range to split:
- *
- * 0 -> 32768(consumer-2) -> 65536(consumer-1)
- *
- * While a consumer removed, The range for this consumer will be taken over
- * by other consumer, consumer-2 removed:
- *
- * 0 -> 65536(consumer-1)
- *
- * In this approach use skip list map to maintain the hash range and consumers.
- *
- * Select consumer will return the ceiling key of message key hashcode % range size.
- *
+ * The implementation uses consistent hashing to evenly split, the
+ * number of keys assigned to each consumer.
*/
public class HashRangeAutoSplitStickyKeyConsumerSelector implements StickyKeyConsumerSelector {
- private final int rangeSize;
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
- private final ConcurrentSkipListMap<Integer, Consumer> rangeMap;
- private final Map<Consumer, Integer> consumerRange;
+ // Consistent-Hash ring
+ private final NavigableMap<Integer, Consumer> hashRing;
- public HashRangeAutoSplitStickyKeyConsumerSelector() {
- this(DEFAULT_RANGE_SIZE);
- }
+ private final int numberOfPOints;
- public HashRangeAutoSplitStickyKeyConsumerSelector(int rangeSize) {
- if (rangeSize < 2) {
- throw new IllegalArgumentException("range size must greater than 2");
- }
- if (!is2Power(rangeSize)) {
- throw new IllegalArgumentException("range size must be nth power with 2");
- }
- this.rangeMap = new ConcurrentSkipListMap<>();
- this.consumerRange = new HashMap<>();
- this.rangeSize = rangeSize;
+ public HashRangeAutoSplitStickyKeyConsumerSelector(int numberOfPOints) {
+ this.hashRing = new TreeMap<>();
+ this.numberOfPOints = numberOfPOints;
}
@Override
- public synchronized void addConsumer(Consumer consumer) throws ConsumerAssignException {
- if (rangeMap.size() == 0) {
- rangeMap.put(rangeSize, consumer);
- consumerRange.put(consumer, rangeSize);
- } else {
- splitRange(findBiggestRange(), consumer);
+ public void addConsumer(Consumer consumer) throws ConsumerAssignException {
+ rwLock.writeLock().lock();
+ try {
+ // Insert multiple points on the hash ring for every consumer
+ // The points are deterministically added based on the hash of the consumer name
+ for (int i = 0; i < numberOfPOints; i++) {
Review comment:
"numberOfPoints"
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie merged pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
sijie merged pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-636163229
@sijie PTAL again
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618176136
> If users want to more evenly distributed, they will increase the points. I'm worried about too many consumers(e.g. 5000 consumers) may result in too many nodes in the TreeMap.
Regarding many consumers, sure the default of 100 points could be overkill, since we'd be fine with 1 point per consumer. It would be easy though to dynamically change the number of points in the ring based on the number of consumers. The only part that we need to be careful during this operation is the same as described in #6554, maintaining the ordering after the hash ranges get readjusted.
I'd say to start with current simple approach, and make it dynamic if it becomes a pain point.
> I'm not sure if this will require more CPU resources, related to [#6647 (comment)](https://github.com/apache/pulsar/pull/6647#issuecomment-618110187). I will test it later.
I have a 2nd PR ready (based on top of this one) to address that and other issues on the dispatcher code.
> Or, can we add a configuration in the broker.conf? so that we don't need to introduce more options at the client-side. I met some users who already use Key_Shared subscription in production. I think we'd better give the user a transition time and we can deprecate the current implementation, change the default value in the broker.conf and finally delete the source code.
The code for the consumer selection is pretty much self-contained and a pretty basic implementation of consistent hashing.
For what is worth, this new implementation is also already running in production without issues.
That also doesn't remove the fact that the current implementation can easily lead to very imbalanced traffic distribution. In fact, the very first user we got to use key-shared feature got bit by this. That should be regarded as a bug in any case.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] sijie commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
sijie commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-619788403
> the original question on having a configuration switch.
Yes. I think the question is to have a configuration switch to allow people to deprecate the current implementation and opt-in the new implementation.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413411500
##########
File path: pulsar-broker/src/test/java/org/apache/pulsar/client/api/NonPersistentKeySharedSubscriptionTest.java
##########
@@ -1,482 +0,0 @@
-/**
Review comment:
I've removed the duplicated code for the non-persistent tests. Instead I've added a matrix of tests at https://github.com/apache/pulsar/pull/6791/files#diff-57c959d83bcb0f61a1de2014e9ea8a30R63
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] merlimat commented on pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
merlimat commented on pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-635000206
@sijie @codelipenghui Added configuration switch for consistent hashing, by default turned off. PTAL.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] jerrypeng commented on a change in pull request #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
jerrypeng commented on a change in pull request #6791:
URL: https://github.com/apache/pulsar/pull/6791#discussion_r413416364
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/HashRangeAutoSplitStickyKeyConsumerSelector.java
##########
@@ -18,86 +18,64 @@
*/
package org.apache.pulsar.broker.service;
-import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
-import org.apache.pulsar.common.util.Murmur3_32Hash;
-
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Map.Entry;
-import java.util.concurrent.ConcurrentSkipListMap;
+import java.util.NavigableMap;
+import java.util.TreeMap;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+
+import org.apache.pulsar.broker.service.BrokerServiceException.ConsumerAssignException;
+import org.apache.pulsar.common.util.Murmur3_32Hash;
/**
* This is a consumer selector based fixed hash range.
*
- * 1.Each consumer serves a fixed range of hash value
- * 2.The whole range of hash value could be covered by all the consumers.
- * 3.Once a consumer is removed, the left consumers could still serve the whole range.
- *
- * Initializing with a fixed hash range, by default 2 << 5.
- * First consumer added, hash range looks like:
- *
- * 0 -> 65536(consumer-1)
- *
- * Second consumer added, will find a biggest range to split:
- *
- * 0 -> 32768(consumer-2) -> 65536(consumer-1)
- *
- * While a consumer removed, The range for this consumer will be taken over
- * by other consumer, consumer-2 removed:
- *
- * 0 -> 65536(consumer-1)
- *
- * In this approach use skip list map to maintain the hash range and consumers.
- *
- * Select consumer will return the ceiling key of message key hashcode % range size.
- *
+ * The implementation uses consistent hashing to evenly split, the
+ * number of keys assigned to each consumer.
*/
public class HashRangeAutoSplitStickyKeyConsumerSelector implements StickyKeyConsumerSelector {
- private final int rangeSize;
+ private final ReadWriteLock rwLock = new ReentrantReadWriteLock();
- private final ConcurrentSkipListMap<Integer, Consumer> rangeMap;
- private final Map<Consumer, Integer> consumerRange;
+ // Consistent-Hash ring
+ private final NavigableMap<Integer, Consumer> hashRing;
- public HashRangeAutoSplitStickyKeyConsumerSelector() {
- this(DEFAULT_RANGE_SIZE);
- }
+ private final int numberOfPOints;
- public HashRangeAutoSplitStickyKeyConsumerSelector(int rangeSize) {
- if (rangeSize < 2) {
- throw new IllegalArgumentException("range size must greater than 2");
- }
- if (!is2Power(rangeSize)) {
- throw new IllegalArgumentException("range size must be nth power with 2");
- }
- this.rangeMap = new ConcurrentSkipListMap<>();
- this.consumerRange = new HashMap<>();
- this.rangeSize = rangeSize;
+ public HashRangeAutoSplitStickyKeyConsumerSelector(int numberOfPOints) {
+ this.hashRing = new TreeMap<>();
+ this.numberOfPOints = numberOfPOints;
}
@Override
- public synchronized void addConsumer(Consumer consumer) throws ConsumerAssignException {
- if (rangeMap.size() == 0) {
- rangeMap.put(rangeSize, consumer);
- consumerRange.put(consumer, rangeSize);
- } else {
- splitRange(findBiggestRange(), consumer);
+ public void addConsumer(Consumer consumer) throws ConsumerAssignException {
+ rwLock.writeLock().lock();
+ try {
+ // Insert multiple points on the hash ring for every consumer
+ // The points are deterministically added based on the hash of the consumer name
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.put(hash, consumer);
+ }
+ } finally {
+ rwLock.writeLock().unlock();
}
}
@Override
- public synchronized void removeConsumer(Consumer consumer) {
- Integer removeRange = consumerRange.remove(consumer);
- if (removeRange != null) {
- if (removeRange == rangeSize && rangeMap.size() > 1) {
- Map.Entry<Integer, Consumer> lowerEntry = rangeMap.lowerEntry(removeRange);
- rangeMap.put(removeRange, lowerEntry.getValue());
- rangeMap.remove(lowerEntry.getKey());
- consumerRange.put(lowerEntry.getValue(), removeRange);
- } else {
- rangeMap.remove(removeRange);
+ public void removeConsumer(Consumer consumer) {
+ rwLock.writeLock().lock();
+ try {
+ // Remove all the points that were added for this consumer
+ for (int i = 0; i < numberOfPOints; i++) {
+ String key = consumer.consumerName() + i;
+ int hash = Murmur3_32Hash.getInstance().makeHash(key.getBytes());
+ hashRing.remove(hash, consumer);
Review comment:
I see thanks for the explanation
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] codelipenghui commented on issue #6791: Use consistent hashing in KeyShared distribution
Posted by GitBox <gi...@apache.org>.
codelipenghui commented on issue #6791:
URL: https://github.com/apache/pulsar/pull/6791#issuecomment-618135261
> AUTO_SPLIT and CONSISTENT_HASHING are essentially 2 different implementations for the same stated behavior. In my view, we should just give the user one good default option. That option should work across all corner cases.
Ok, make sense to me.
If users want to more evenly distributed, they will increase the points. I'm worried about too many consumers(e.g. 5000 consumers) may result in too many nodes in the TreeMap. I'm not sure if this will require more CPU resources, related to https://github.com/apache/pulsar/pull/6647#issuecomment-618110187. I will test it later.
Or, can we add a configuration in the broker.conf? so that we don't need to introduce more options at the client-side. I met some users who already use Key_Shared subscription in production. I think we'd better give the user a transition time and we can deprecate the current implementation, change the default value in the broker.conf and finally delete the source code.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org