[fix][broker]Fix failover/exclusive consumer cumulate ack not remove msg from UnAckedMessageTracker #18407
Conversation
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Technoboy-
added
type/bug
Codecov Report
@@ Coverage Diff @@
## master #18407 +/- ##
============================================
- Coverage 47.67% 46.60% -1.08%
+ Complexity 9333 7687 -1646
============================================
Files 618 463 -155
Lines 58586 51659 -6927
Branches 6098 5514 -584
============================================
- Hits 27933 24077 -3856
+ Misses 27623 24779 -2844
+ Partials 3030 2803 -227
Flags with carried forward coverage won't be shown. Click here to find out more.
|
coderzc
reviewed
Nov 10, 2022
| @@ -458,7 +458,8 @@ protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ack | |||
| Consumer individualConsumer = consumers.get(topicMessageId.getTopicPartitionName()); | |||
| if (individualConsumer != null) { | |||
| MessageId innerId = topicMessageId.getInnerMessageId(); | |||
| return individualConsumer.acknowledgeCumulativeAsync(innerId); | |||
| return individualConsumer.acknowledgeCumulativeAsync(innerId) | |||
| .thenAccept(__ -> unAckedMessageTracker.remove(topicMessageId)); | |||
There was a problem hiding this comment.
Suggested change
| .thenAccept(__ -> unAckedMessageTracker.remove(topicMessageId)); | |
| .thenAccept(__ -> unAckedMessageTracker.removeMessagesTill(topicMessageId)); |
poorbarcode
approved these changes
Nov 10, 2022
| .acknowledgmentGroupTime(100, TimeUnit.MILLISECONDS) | ||
| .subscriptionInitialPosition(SubscriptionInitialPosition.Latest) | ||
| .subscribe(); | ||
| final MessageId msgId = producer.send("1".getBytes()); |
There was a problem hiding this comment.
Should we send more than one messages and ack the last one to ensure unAckedMessageTracker.removeMessagesTill works?
coderzc
approved these changes
Nov 11, 2022
There was a problem hiding this comment.
What about doReconsumeLater method? and also, I'm wondering if this can work properly with the partitioned topic.
Consider this condition:
- We have 4 partitioned topics. the ledgers are
ledger id: 1-> partition 1, 'ledger id: 2' ->partition 2... - If one of the partitions, for example, partition-2 create a new ledger that id=5.
- At that time, the producer sends the message in round-robin mode. So, the message id sequence may be like [1:0, 1:1, 5:0, 5:1, 5:2, 5:3, 3:0, 3:1, 4:1, 4:2] in the consumer incoming queue.
- if we try to cumulative ack message 5:0, we will clear [3:0, 3:1, 4:1] in the
UnAckedMessageTracker.
I'm not sure if I missing some other logic to avoid this condition. but I think I have to mention it here.
mattisonchao
requested review from
codelipenghui,
congbobo184,
lhotari and
michaeljmarshall
Comment on lines
461
to
462
| return individualConsumer.acknowledgeCumulativeAsync(innerId) | ||
| .thenAccept(__ -> unAckedMessageTracker.removeMessagesTill(topicMessageId)); |
There was a problem hiding this comment.
The unAckedMessageTracker will have the message IDs from all partitions. If we use removeMessagesTill here, the message acknowledgment for partition-0 will also remove the message IDs of other partitions, right?
Technoboy-
force-pushed
the
fix-cumulate-ack
branch
from
9c168e9
to
326b1cf
Compare
congbobo184
reviewed
Dec 9, 2022
| @@ -461,7 +461,8 @@ protected CompletableFuture<Void> doAcknowledge(MessageId messageId, AckType ack | |||
| Consumer individualConsumer = consumers.get(topicMessageId.getTopicPartitionName()); | |||
| if (individualConsumer != null) { | |||
| MessageId innerId = topicMessageId.getInnerMessageId(); | |||
| return individualConsumer.acknowledgeCumulativeAsync(innerId); | |||
| return individualConsumer.acknowledgeCumulativeAsync(innerId) | |||
| .thenAccept(__ -> unAckedMessageTracker.remove(topicMessageId)); | |||
There was a problem hiding this comment.
cumulativeAck, not every message will invoke, so use remove can't remove all. if use removeMessagsTill(), all partitioned topics will be removed, so its better to find a better way to handle this case
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Motivation
If the consumer subscription type is
Failover, orExclusive, when do cumulate ack, the msg is not removed from UnAckedMessageTracker, causing the msg to be triggered redeliver by ack timeout.,
Modifications
Remove the msg from UnAckedMessageTracker when complete
acknowledgeCumulativeAsyncDocumentation
docdoc-requireddoc-not-neededdoc-complete