[fix][broker] Fix dispatch duplicated messages with Exclusive mode.
#17237
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
mattisonchao
commented
Aug 23, 2022
...ache/pulsar/broker/service/persistent/PersistentStreamingDispatcherSingleActiveConsumer.java
Show resolved
Hide resolved
...va/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumer.java
Outdated
Show resolved
Hide resolved
|
I will add more information later |
mattisonchao
commented
Aug 23, 2022
...er/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
Outdated
Show resolved
Hide resolved
|
Because branch-2.10 also has this problem. I don't think it's a 2.11 release blocker.
|
From this log, we can know |
mattisonchao
requested review from
codelipenghui,
lhotari,
Technoboy-,
eolivelli,
nicoloboschi,
congbobo184 and
michaeljmarshall
codelipenghui
approved these changes
Aug 23, 2022
...er/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
Outdated
Show resolved
Hide resolved
...er/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
Show resolved
Hide resolved
...er/src/test/java/org/apache/pulsar/client/api/SubscriptionMessageDispatchThrottlingTest.java
Outdated
Show resolved
Hide resolved
|
@mattisonchao Do you know which PR introduced this issue? |
|
This is very interesting. |
| @@ -332,12 +332,17 @@ public void redeliverUnacknowledgedMessages(Consumer consumer, List<PositionImpl | |||
| } | |||
|
|
|||
| @Override | |||
| protected void readMoreEntries(Consumer consumer) { | |||
| protected synchronized void readMoreEntries(Consumer consumer) { | |||
There was a problem hiding this comment.
It seems like we might not want to acquire the lock before the consumer null check.
There was a problem hiding this comment.
Modified the scope of the lock
| // consumer can be null when all consumers are disconnected from broker. | ||
| // so skip reading more entries if currently there is no active consumer. | ||
| if (null == consumer) { | ||
| return; | ||
| } | ||
| if (havePendingRead) { |
There was a problem hiding this comment.
I see several parts of the code that call this readMoreEntries method after verifying that havePendingRead is true. I haven't yet verified the details of each call, but it seems like we should clean up those guards if we are moving the logic into this method.
There was a problem hiding this comment.
I think we should move the havePendingRead check to the readMoreEntries method to avoid judgement everywhere.
Except for some judgments that have logical implications, I have changed them all.
Please take a look.
mattisonchao
commented
Aug 24, 2022
| Pair<Integer, Long> calculateResult = calculateToRead(consumer); | ||
| int messagesToRead = calculateResult.getLeft(); | ||
| long bytesToRead = calculateResult.getRight(); | ||
| synchronized (this) { |
There was a problem hiding this comment.
Move lock here to avoid invoking useless calculateToRead many times.
mattisonchao
commented
Aug 24, 2022
| + " havePendingRead {}", | ||
| topic.getName(), currentConsumer, havePendingRead); | ||
| } | ||
| if (isRescheduleReadInProgress.compareAndSet(false, true)) { |
There was a problem hiding this comment.
We have to avoid reScheduleRead many times, see #16241
|
After the test, I think the problem was caused by we probably invoking So, we may read the same position many times. Even though I fix the |
|
Hi @eolivelli
When I wanted to fix the flaky test "SubscriptionMessageDispatchThrottlingTest", I found that one time we received more messages than we sent. So, I added |
eolivelli
changed the title
Exclusive mode.Exclusive mode.
mattisonchao
requested review from
codelipenghui,
nicoloboschi and
michaeljmarshall
lhotari
reviewed
Aug 26, 2022
Comment on lines
+300
to
+301
| // If this assertion failed, please alert we may have some regression cause message dispatch was duplicated. | ||
| Assert.assertEquals(totalReceived.get(), numProducedMessages, 10); |
There was a problem hiding this comment.
This is flaky. since the comment is "please alert", I'm alerting. :)
There was a problem hiding this comment.
I was running it locally.
java.lang.AssertionError:
Expected :30.0
Actual :44.0
There was a problem hiding this comment.
I will try to verify it. Maybe I lost something. very thanks to you!
Nicklee007
pushed a commit
to Nicklee007/pulsar
that referenced
this pull request
Aug 29, 2022
|
We meet same issue here on flink-connector-pulsar. |
liangyepianzhou
pushed a commit
that referenced
this pull request
Dec 6, 2022
nicoloboschi
pushed a commit
to datastax/pulsar
that referenced
this pull request
Dec 7, 2022
…apache#17237) (cherry picked from commit 0517423) (cherry picked from commit 390a4ed)
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
When invoking
readMoreEntrieswe have to check if thehavePendingReadstate istrueto avoid reading the same position many times in the race condition.You can add
invocationCounton theTestannotation to call the new test multiple times to understand the problem.The relative logs:
Modifications
synchronizedkeyword andhavePendingReadstate to avoid read same position many times.havePendingReadcheck.Verifying this change
Documentation
Check the box below or label this PR directly.
Need to update docs?
doc-required(Your PR needs to update docs and you will update later)
doc-not-needed(Please explain why)
doc(Your PR contains doc changes)
doc-complete(Docs have been already added)