Issue 16802: fix Repeated messages of shared dispatcher #16812
Merged
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
Changes: sendMessagesToConsumer must block reads of new entries otherwise we will end up in reading duplicates
eolivelli
added
the
release/blocker
| break; | ||
| sendInProgress = true; | ||
| try { | ||
| if (needTrimAckedMessages()) { |
There was a problem hiding this comment.
Could you move this try block to another method to avoid so many code changes? For example,
private void trySendMessagesToConsumers(ReadType readType, List<Entry> entries) {
/* ... */
}
protected synchronized void sendMessagesToConsumers(ReadType readType, List<Entry> entries) {
sendInProgress = true;
try {
trySendMessagesToConsumers(readType, entries);
} finally {
sendInProgress = false;
}
readMoreEntries();
}In addition, the following code in your original code can be removed because in the finally block sendInProgress would be false, then readMoreEntries will be called after that.
if (entriesToDispatch == 0) {
sendInProgress = false;
readMoreEntries();
return;
}
This was referenced Jul 27, 2022
BewareMyPower
approved these changes
Jul 27, 2022
|
|
||
| protected synchronized void trySendMessagesToConsumers(ReadType readType, List<Entry> entries) { |
There was a problem hiding this comment.
The PersistentStickyKeyDispatcherMultipleConsumers should also override this method.
There was a problem hiding this comment.
you are right, thanks for the heads up
I will also add a comment for future people who will put their hands here
| @@ -278,7 +278,6 @@ public void testDelayedDeliveryWithMultipleConcurrentReadEntries() | |||
| for (int i = 0; i < N; i++) { | |||
| msg = consumer.receive(10, TimeUnit.SECONDS); | |||
| receivedMsgs.add(msg.getValue()); | |||
| consumer.acknowledge(msg); | |||
There was a problem hiding this comment.
Why need to remove this line?
There was a problem hiding this comment.
I added that line in the PR that created this problem.
this line made the test pass,
with this "fix" now the test still passes
this test is very interesting because it runs many readMoreEntries() in a separate thread, simulating the mess that happens inside the broker
it is important to revert this change to the test
codelipenghui
approved these changes
Jul 27, 2022
|
/pulsarbot rerun-failure-checks |
|
Please check the failed test |
|
@codelipenghui I am looking into the failed test. |
|
@codelipenghui I have fixed the test, there was a problem in the KEY_SHARED dispatcher |
codelipenghui
approved these changes
Jul 28, 2022
Technoboy-
approved these changes
Jul 28, 2022
|
@codelipenghui @BewareMyPower I have merged the PR |
1 task
Gleiphir2769
pushed a commit
to Gleiphir2769/pulsar
that referenced
this pull request
Aug 4, 2022
2 tasks
2 tasks
1 task
4 tasks
labuladong
reviewed
Nov 6, 2022
| @@ -89,7 +89,7 @@ public class PersistentDispatcherMultipleConsumers extends AbstractDispatcherMul | |||
| protected volatile PositionImpl minReplayedPosition = null; | |||
| protected boolean shouldRewindBeforeReadingOrReplaying = false; | |||
| protected final String name; | |||
|
|
|||
| protected boolean sendInProgress; | |||
There was a problem hiding this comment.
I think this flag should be volatile?
There was a problem hiding this comment.
Itbis guarded by synchronized blocks. No need to makenit volatile
There was a problem hiding this comment.
I misunderstood something, you're right.
liangyepianzhou
pushed a commit
that referenced
this pull request
Jul 5, 2023
(cherry picked from commit 825b68d)
liangyepianzhou
added a commit
that referenced
this pull request
Jul 5, 2023
…sages of shared dispatcher (#16812)
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.
Changes: sendMessagesToConsumer must block reads of new entries otherwise we will end up in reading duplicates
Fixes #16802 #16693
Motivation
As described in #16802 it may happen that multiple calls to
readMoreEntriesare served before the execution ofsendMessagesToConsumerUnfortunately there is no control over the execution of
readMoreEntries, it may happen in many different threadpoolsModifications
Add a flag to prevent concurrent execution of
sendMessagesToConsumerwithreadMoreEntries.Revert aa change to a test modified in #16802 to make the test pass
Verifying this change
This change is already covered by existing tests
doc-not-needed