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 2022/06/07 15:04:06 UTC
[GitHub] [pulsar] coderzc opened a new pull request, #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
coderzc opened a new pull request, #15963:
URL: https://github.com/apache/pulsar/pull/15963
Fixes #15898
### Motivation
Now, when can‘t match the consumer message will add the entry to the list of messages to be redelivered, this will lead to other consumers may not be able to receive a message.
### Modifications
Add backoff for the redelivering message when the message can't match a consumer
### Verifying this change
- [x] Make sure that the change passes the CI checks.
### Documentation
Check the box below or label this PR directly.
Need to update docs?
- [x] `doc-not-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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] coderzc commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r893034194
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
> Seems this config not work.
I removed the config
--
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
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892337710
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
Seems this config not work.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
Seems this config not work.
--
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
[GitHub] [pulsar] coderzc commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892543367
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -215,13 +219,18 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
for (Entry entry : entries) {
- int stickyKeyHash = getStickyKeyHash(entry);
+ final int stickyKeyHash = getStickyKeyHash(entry);
+ final long ledgerId = entry.getLedgerId();
+ final long entryId = entry.getEntryId();
Consumer c = selector.select(stickyKeyHash);
if (c != null) {
groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
} else {
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ topic.getBrokerService().executor().schedule(() -> {
Review Comment:
> Is there any difference between scheduled or directly executing the method `addMessageToReplay`?
>
> Do we need to reset the backoff param?
If not scheduled execution, the failed message will be redelivery constantly, which will lead to other messages having no chance to be sent.
I think can rest the backoff when matching a consumer
--
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
[GitHub] [pulsar] coderzc commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892543367
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -215,13 +219,18 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
for (Entry entry : entries) {
- int stickyKeyHash = getStickyKeyHash(entry);
+ final int stickyKeyHash = getStickyKeyHash(entry);
+ final long ledgerId = entry.getLedgerId();
+ final long entryId = entry.getEntryId();
Consumer c = selector.select(stickyKeyHash);
if (c != null) {
groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
} else {
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ topic.getBrokerService().executor().schedule(() -> {
Review Comment:
> Is there any difference between scheduled or directly executing the method `addMessageToReplay`?
>
> Do we need to reset the backoff param?
If not scheduled execution, the failed message will be redelivery constantly, which will lead to other messages having no chance to be sent.
--
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
[GitHub] [pulsar] coderzc commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892543367
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -215,13 +219,18 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
for (Entry entry : entries) {
- int stickyKeyHash = getStickyKeyHash(entry);
+ final int stickyKeyHash = getStickyKeyHash(entry);
+ final long ledgerId = entry.getLedgerId();
+ final long entryId = entry.getEntryId();
Consumer c = selector.select(stickyKeyHash);
if (c != null) {
groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
} else {
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ topic.getBrokerService().executor().schedule(() -> {
Review Comment:
> Is there any difference between scheduled or directly executing the method `addMessageToReplay`?
>
> Do we need to reset the backoff param?
If not scheduled execution, the failed message will be redelivery constantly, which will lead to other messages having no chance to be sent.
I think can reset the backoff when matching a consumer
--
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
[GitHub] [pulsar] Technoboy- commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
Technoboy- commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892338294
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
You need to stop the broker and then start again.
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
You need to stop the broker and then start 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.
To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] github-actions[bot] commented on pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
github-actions[bot] commented on PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#issuecomment-1193229752
The pr had no activity for 30 days, mark with Stale label.
--
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
[GitHub] [pulsar] coderzc closed pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc closed pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
URL: https://github.com/apache/pulsar/pull/15963
--
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
[GitHub] [pulsar] coderzc commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
coderzc commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r893034194
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
> Seems this config not work.
I removed the config, it already be deprecated
##########
pulsar-broker/src/test/java/org/apache/pulsar/client/api/KeySharedSubscriptionTest.java:
##########
@@ -363,6 +363,51 @@ public void testNonKeySendAndReceiveWithHashRangeExclusiveStickyKeyConsumerSelec
receiveAndCheck(checkList);
}
+ @Test(dataProvider = "batch")
+ public void testKeySendAndReceiveWithHashRangeExclusiveStickyKeyUnMatchConsumer(boolean enableBatch)
+ throws PulsarClientException, InterruptedException {
+ this.conf.setSubscriptionKeySharedEnable(true);
Review Comment:
> Seems this config not work.
I removed the config, it has already be deprecated
--
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
[GitHub] [pulsar] gaoran10 commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
gaoran10 commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r892437081
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -215,13 +219,18 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
for (Entry entry : entries) {
- int stickyKeyHash = getStickyKeyHash(entry);
+ final int stickyKeyHash = getStickyKeyHash(entry);
+ final long ledgerId = entry.getLedgerId();
+ final long entryId = entry.getEntryId();
Consumer c = selector.select(stickyKeyHash);
if (c != null) {
groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
} else {
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ topic.getBrokerService().executor().schedule(() -> {
Review Comment:
Is there any difference between scheduled or directly executing the method `addMessageToReplay`?
Do we need to reset the backoff param?
--
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
[GitHub] [pulsar] Jason918 commented on a diff in pull request #15963: [fix][broker] KeyShared stickyHashRange subscription: Fix not be able to receive a message when send a can‘t match consumer message
Posted by GitBox <gi...@apache.org>.
Jason918 commented on code in PR #15963:
URL: https://github.com/apache/pulsar/pull/15963#discussion_r895289610
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentStickyKeyDispatcherMultipleConsumers.java:
##########
@@ -215,13 +219,19 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
final Map<Consumer, Set<Integer>> consumerStickyKeyHashesMap = new HashMap<>();
for (Entry entry : entries) {
- int stickyKeyHash = getStickyKeyHash(entry);
+ final int stickyKeyHash = getStickyKeyHash(entry);
+ final long ledgerId = entry.getLedgerId();
+ final long entryId = entry.getEntryId();
Consumer c = selector.select(stickyKeyHash);
if (c != null) {
groupedEntries.computeIfAbsent(c, k -> new ArrayList<>()).add(entry);
consumerStickyKeyHashesMap.computeIfAbsent(c, k -> new HashSet<>()).add(stickyKeyHash);
+ unMatchConsumerBackoff.reset();
} else {
- addMessageToReplay(entry.getLedgerId(), entry.getEntryId(), stickyKeyHash);
+ topic.getBrokerService().executor().schedule(() -> {
+ addMessageToReplay(ledgerId, entryId, stickyKeyHash);
+ }, unMatchConsumerBackoff.next(), TimeUnit.MILLISECONDS);
Review Comment:
`unMatchConsumerBackoff.next()` seems odd here.
Consider the case when all entry in `entries` have the same `stickyKeyHash`, the delay time increases in the same batch entries.
--
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