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 2021/04/28 08:15:57 UTC
[GitHub] [pulsar] rdhabalia opened a new pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
rdhabalia opened a new pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417
### Motivation
This will fix #6054 , It happens when a topic which has batch messages, subscription has multiple consumers and if some of those consumers' receive more messages than available-permits (due to unequal batch distribution which will be fixed by #7266) and if they are slow then totalAvailablePermits become -ve for that subscription and broker will not dispatch messages to the consumes which still have permits.
```
# slow consumer with -ve permit that stopped consuming messages
"msgRateOut": 0,
"msgThroughputOut": 0,
"bytesOutCounter": 2993076140,
"msgOutCounter": 325043,
"msgRateRedeliver": 0,
"chuckedMessageRate": 0,
"chunkedMessageRate": 0,
"consumerName": "bee1d",
"availablePermits": -43,
"unackedMessages": 1089,
"msgRateOut": 0,
"msgThroughputOut": 0,
"bytesOutCounter": 3021328186,
"msgOutCounter": 326508,
"msgRateRedeliver": 0,
"chuckedMessageRate": 0,
"chunkedMessageRate": 0,
"consumerName": "57f72",
"availablePermits": -8,
"unackedMessages": 1233,
# consumer with permit which doesn't receive message from broker
"msgRateOut": 0,
"msgThroughputOut": 0,
"bytesOutCounter": 2960672336,
"msgOutCounter": 321930,
"msgRateRedeliver": 0,
"chuckedMessageRate": 0,
"chunkedMessageRate": 0,
"consumerName": "fdc38",
"availablePermits": 70,
"unackedMessages": 1213,
"avgMessagesPerEntry": 11,
"blockedConsumerOnUnackedMsgs": false,
"lastAckedTimestamp": 1619577828969,
"lastConsumedTimestamp": 1619577832806,
"metadata": {
"instance_id": "0",
"application": "pulsar-function",
"instance_hostname": "fab08.xyz.com",
"id": "amplitude/processing/siege-filter"
},
```
### Modification
The broker should dispatch messages if any of the consumers of subscription has permits available so, stuck/slow consumer doesn't impact other good consumers.
### Result
It should fix #6054
cc @devinbost
--
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] lhotari commented on a change in pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#discussion_r622159850
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
##########
@@ -480,7 +481,9 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
long totalMessagesSent = 0;
long totalBytesSent = 0;
- while (entriesToDispatch > 0 && totalAvailablePermits > 0 && isAtleastOneConsumerAvailable()) {
+ int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits();
+ int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits);
+ while (entriesToDispatch > 0 && currentTotalAvailablePermits > 0 && firstAvailableConsumerPermits > 0) {
Review comment:
isn't there a difference in behavior here now that `firstAvailableConsumerPermits > 0` is evaluated once before the while loop and previously `isAtleastOneConsumerAvailable()` would get evaluated each time the while loop evaluates the condition?
--
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] devinbost commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
devinbost commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-830473812
BTW, to any reviewers, I have tested this branch on a custom build of Pulsar, and it works as expected.
--
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] rdhabalia merged pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
rdhabalia merged pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417
--
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] rdhabalia commented on a change in pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
rdhabalia commented on a change in pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#discussion_r622320825
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
##########
@@ -480,7 +481,9 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
long totalMessagesSent = 0;
long totalBytesSent = 0;
- while (entriesToDispatch > 0 && totalAvailablePermits > 0 && isAtleastOneConsumerAvailable()) {
+ int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits();
+ int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits);
+ while (entriesToDispatch > 0 && currentTotalAvailablePermits > 0 && firstAvailableConsumerPermits > 0) {
Review comment:
thanks for catching it. yes, I missed to push the commit. fixed 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.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [pulsar] devinbost commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
devinbost commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-830469650
@rdhabalia Please update the description of this issue so it doesn't automatically close #6054 since we determined that this PR is only a partial fix for that issue. I'll update the issue with our latest findings.
--
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] lhotari commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
lhotari commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-828435616
Thank you @rdhabalia for working on a fix.
Would you be able to review #10413 since that is also working towards fixing issues in the same area?
--
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] devinbost commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
devinbost commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-828783156
/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] devinbost commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
devinbost commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-828648766
/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] lhotari commented on a change in pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
lhotari commented on a change in pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#discussion_r622160806
##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java
##########
@@ -480,7 +481,9 @@ protected void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
long totalMessagesSent = 0;
long totalBytesSent = 0;
- while (entriesToDispatch > 0 && totalAvailablePermits > 0 && isAtleastOneConsumerAvailable()) {
+ int firstAvailableConsumerPermits = getFirstAvailableConsumerPermits();
+ int currentTotalAvailablePermits = Math.max(totalAvailablePermits, firstAvailableConsumerPermits);
+ while (entriesToDispatch > 0 && currentTotalAvailablePermits > 0 && firstAvailableConsumerPermits > 0) {
Review comment:
perhaps something like `(firstAvailableConsumerPermits > 0 || isAtleastOneConsumerAvailable())` ?
--
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] devinbost commented on pull request #10417: [pulsar-broker] Dispatch messaages to consumer with permits
Posted by GitBox <gi...@apache.org>.
devinbost commented on pull request #10417:
URL: https://github.com/apache/pulsar/pull/10417#issuecomment-830469701
/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