You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by "lhotari (via GitHub)" <gi...@apache.org> on 2024/10/11 15:07:41 UTC
[PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
lhotari opened a new pull request, #23441:
URL: https://github.com/apache/pulsar/pull/23441
Fixes #23439
### Motivation
When a consumer joins or leaves, the hash ranges from existing consumers shouldn't move from one to another.
### Modifications
- Revisit adding the to hash ring so that existing consumers aren't replaced when there are collisions.
- Add a test case which ensures this
### Documentation
<!-- DO NOT REMOVE THIS SECTION. CHECK THE PROPER BOX ONLY. -->
- [ ] `doc` <!-- Your PR contains doc changes. -->
- [ ] `doc-required` <!-- Your PR changes impact docs and you will update later -->
- [x] `doc-not-needed` <!-- Your PR changes do not impact docs -->
- [ ] `doc-complete` <!-- Docs have been already added -->
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797156817
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
One problem is that the hash points don't get restored for a collided consumer if an earlier consumer leaves. A solution where there's a list and the first entry is used would be a solution to this. That's one reason to restore the list in the hash ring entry. I'll create an issue to track 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "codecov-commenter (via GitHub)" <gi...@apache.org>.
codecov-commenter commented on PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#issuecomment-2407821569
## [Codecov](https://app.codecov.io/gh/apache/pulsar/pull/23441?dropdown=coverage&src=pr&el=h1&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) Report
Attention: Patch coverage is `56.25000%` with `7 lines` in your changes missing coverage. Please review.
> Project coverage is 74.23%. Comparing base [(`bbc6224`)](https://app.codecov.io/gh/apache/pulsar/commit/bbc62245c5ddba1de4b1e7cee4ab49334bc36277?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) to head [(`cc54bf8`)](https://app.codecov.io/gh/apache/pulsar/commit/cc54bf8983d9a2cb25df359e9d0ea5ca13d9f818?dropdown=coverage&el=desc&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache).
> Report is 666 commits behind head on master.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23441?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Patch % | Lines |
|---|---|---|
| [...ce/ConsistentHashingStickyKeyConsumerSelector.java](https://app.codecov.io/gh/apache/pulsar/pull/23441?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FConsistentHashingStickyKeyConsumerSelector.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0NvbnNpc3RlbnRIYXNoaW5nU3RpY2t5S2V5Q29uc3VtZXJTZWxlY3Rvci5qYXZh) | 53.33% | [5 Missing and 2 partials :warning: ](https://app.codecov.io/gh/apache/pulsar/pull/23441?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) |
<details><summary>Additional details and impacted files</summary>
[](https://app.codecov.io/gh/apache/pulsar/pull/23441?src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
```diff
@@ Coverage Diff @@
## master #23441 +/- ##
============================================
+ Coverage 73.57% 74.23% +0.66%
- Complexity 32624 34431 +1807
============================================
Files 1877 1953 +76
Lines 139502 147189 +7687
Branches 15299 16206 +907
============================================
+ Hits 102638 109272 +6634
- Misses 28908 29472 +564
- Partials 7956 8445 +489
```
| [Flag](https://app.codecov.io/gh/apache/pulsar/pull/23441/flags?src=pr&el=flags&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [inttests](https://app.codecov.io/gh/apache/pulsar/pull/23441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `27.65% <56.25%> (+3.07%)` | :arrow_up: |
| [systests](https://app.codecov.io/gh/apache/pulsar/pull/23441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `24.35% <37.50%> (+0.03%)` | :arrow_up: |
| [unittests](https://app.codecov.io/gh/apache/pulsar/pull/23441/flags?src=pr&el=flag&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | `73.60% <56.25%> (+0.75%)` | :arrow_up: |
Flags with carried forward coverage won't be shown. [Click here](https://docs.codecov.io/docs/carryforward-flags?utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#carryforward-flags-in-the-pull-request-comment) to find out more.
| [Files with missing lines](https://app.codecov.io/gh/apache/pulsar/pull/23441?dropdown=coverage&src=pr&el=tree&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache) | Coverage Δ | |
|---|---|---|
| [...roker/service/ConsumerHashAssignmentsSnapshot.java](https://app.codecov.io/gh/apache/pulsar/pull/23441?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FConsumerHashAssignmentsSnapshot.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0NvbnN1bWVySGFzaEFzc2lnbm1lbnRzU25hcHNob3QuamF2YQ==) | `83.00% <100.00%> (ø)` | |
| [...ce/ConsistentHashingStickyKeyConsumerSelector.java](https://app.codecov.io/gh/apache/pulsar/pull/23441?src=pr&el=tree&filepath=pulsar-broker%2Fsrc%2Fmain%2Fjava%2Forg%2Fapache%2Fpulsar%2Fbroker%2Fservice%2FConsistentHashingStickyKeyConsumerSelector.java&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache#diff-cHVsc2FyLWJyb2tlci9zcmMvbWFpbi9qYXZhL29yZy9hcGFjaGUvcHVsc2FyL2Jyb2tlci9zZXJ2aWNlL0NvbnNpc3RlbnRIYXNoaW5nU3RpY2t5S2V5Q29uc3VtZXJTZWxlY3Rvci5qYXZh) | `92.15% <53.33%> (-2.08%)` | :arrow_down: |
... and [636 files with indirect coverage changes](https://app.codecov.io/gh/apache/pulsar/pull/23441/indirect-changes?src=pr&el=tree-more&utm_medium=referral&utm_source=github&utm_content=comment&utm_campaign=pr+comments&utm_term=apache)
</details>
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797156817
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
One problem is that the hash points don't get restored if an earlier consumer leaves. A solution where there's a list and the first entry is used would be a solution to this. That's one reason to restore the list in the hash ring entry. I'll create an issue to track 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "equanz (via GitHub)" <gi...@apache.org>.
equanz commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797140960
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
It can be done outside of scope, but I think it would be better if this state could be sent to consumers.
(Consumers can't know there are no hash points, can they?)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari merged PR #23441:
URL: https://github.com/apache/pulsar/pull/23441
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797154889
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
Yes, let's handle it if it becomes a problem. It would only happen due to misconfiguration or when the hash range (size 65535) is completely full. It's hard to see when that would happen in real world use case. I think that the error logging is sufficient for now.
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "equanz (via GitHub)" <gi...@apache.org>.
equanz commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797140960
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
It can be out of scope, but I think it would be better if this state could be sent to consumers.
(Consumers can't know there are no hash points, can they?)
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797160006
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
Issue to track this: #23443 23443
--
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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [improve][broker][PIP-379] Don't replace a consumer when there's a collision [pulsar]
Posted by "lhotari (via GitHub)" <gi...@apache.org>.
lhotari commented on code in PR #23441:
URL: https://github.com/apache/pulsar/pull/23441#discussion_r1797721718
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ConsistentHashingStickyKeyConsumerSelector.java:
##########
@@ -76,18 +78,36 @@ public CompletableFuture<Optional<ImpactedConsumersResult>> addConsumer(Consumer
ConsumerIdentityWrapper consumerIdentityWrapper = new ConsumerIdentityWrapper(consumer);
// Insert multiple points on the hash ring for every consumer
// The points are deterministically added based on the hash of the consumer name
+ int hashPointsAdded = 0;
+ int hashPointCollisions = 0;
for (int i = 0; i < numberOfPoints; i++) {
int consumerNameIndex =
consumerNameIndexTracker.increaseConsumerRefCountAndReturnIndex(consumerIdentityWrapper);
int hash = calculateHashForConsumerAndIndex(consumer, consumerNameIndex, i);
- // When there's a collision, the new consumer will replace the old one.
- // This is a rare case, and it is acceptable to replace the old consumer since there
- // are multiple points for each consumer. This won't affect the overall distribution significantly.
- ConsumerIdentityWrapper removed = hashRing.put(hash, consumerIdentityWrapper);
- if (removed != null) {
- consumerNameIndexTracker.decreaseConsumerRefCount(removed);
+ // When there's a collision, the entry won't be added to the hash ring.
+ // This isn't a problem with the consumerNameIndexTracker solution since the collisions won't align
+ // for all hash ring points when using the same consumer name. This won't affect the overall
+ // distribution significantly when the number of hash ring points is sufficiently large (>100).
+ ConsumerIdentityWrapper existing = hashRing.putIfAbsent(hash, consumerIdentityWrapper);
+ if (existing != null) {
+ hashPointCollisions++;
+ // reduce the ref count which was increased before adding since the consumer was not added
+ consumerNameIndexTracker.decreaseConsumerRefCount(consumerIdentityWrapper);
+ } else {
+ hashPointsAdded++;
}
}
+ if (hashPointsAdded == 0) {
+ log.error("Failed to add consumer '{}' to the hash ring. There were {} collisions. Consider increasing "
+ + "the number of points ({}) per consumer by setting "
+ + "subscriptionKeySharedConsistentHashingReplicaPoints={}",
+ consumer, hashPointCollisions, numberOfPoints,
+ Math.max((int) (numberOfPoints * 1.5d), numberOfPoints + 1));
+ }
Review Comment:
@equanz @merlimat please review #23447, it addresses the remaining 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org