when consumer close and create frequently, delay message will be duplicated#14727
when consumer close and create frequently, delay message will be duplicated#14727leizhiyuan wants to merge 7 commits intoapache:masterfrom
Conversation
|
@leizhiyuan:Thanks for your contribution. For this PR, do we need to update docs? |
|
@leizhiyuan:Thanks for providing doc info! |
...ar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
Outdated
Show resolved
Hide resolved
...ar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
Outdated
Show resolved
Hide resolved
...ar-broker/src/main/java/org/apache/pulsar/broker/delayed/InMemoryDelayedDeliveryTracker.java
Outdated
Show resolved
Hide resolved
codelipenghui
requested review from
315157973,
Jason918,
Technoboy-,
aloyszhang,
hangc0276 and
merlimat
|
Can we add a units test to cover this case? |
|
|
||
| if (dispatcher != null && dispatcher.getConsumers().isEmpty()) { | ||
| // clear delay message avoid duplicate. | ||
| dispatcher.clearDelayedMessages(); |
There was a problem hiding this comment.
There are other places will call cursor.rewind(); in the PersistentDispatcherMultipleConsumers. The cursor rewind will cause to add the duplicated delayed message index, and reset the cursor will also introduce this problem, could you please check them?
| // All consumers got disconnected before the completion of the read operation | ||
| entries.forEach(Entry::release); | ||
| cursor.rewind(); | ||
| clearDelayedMessages(); |
There was a problem hiding this comment.
If clearDelayedMessages() binds with cursor.rewind(), it's better to wrap them into one method.
|
see #14740 |
8 participants
Fixes #14722
(or if this PR is one task of a github issue, please add
Master Issue: #<xyz>to link to the master issue.)Master Issue: #14722
Motivation
Explain here the context, and why you're making that change. What is the problem you're trying to solve.
Modifications
Describe the modifications you've done.
Verifying this change
(Please pick either of the following options)
This change is a trivial rework / code cleanup without any test coverage.
(or)
This change is already covered by existing tests, such as (please describe tests).
(or)
This change added tests and can be verified as follows:
(example:)
Does this pull request potentially affect one of the following parts:
If
yeswas chosen, please highlight the changesDocumentation
Check the box below or label this PR directly (if you have committer privilege).
Need to update docs?
doc-required(If you need help on updating docs, create a doc issue)
no-need-doc(Please explain why)
doc(If this PR contains doc changes)