Fix race condition when calling acknowledgementWasProcessed() #8499
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
sijie
approved these changes
Nov 11, 2020
codelipenghui
added a commit
that referenced
this pull request
Nov 12, 2020
1. Fix race condition when calling `acknowledgementWasProcessed()`. Currently, when the broker received a message acknowledge request, the broker will call ManagedCursor to delete the position by using `cursor.asyncDelete()`. We should call the acknowledgementWasProcessed() in the delete position callback. Otherwise, when the dispatcher for the Key_Shared subscription will get an unchanged mark delete position so that in some case the consumer can't be removed from the `recentJoinedConsumers`. This might cause the message consumption stuck in the Key_Shared subscription. 2. Currently, the consumer removed from the `recentJoinedConsumers` until the mark delete position is greater than the read position that the consumer joined and there are new messages need to deliver to this consumer. But in some case, there are no new messages need to deliver to this consumer, so this consumer can't be moved from the `recentJoinedConsumers`. This will also lead to consumption stuck. (cherry picked from commit 8df7364)
|
cherry-picked to branch-2.6 |
codelipenghui
added a commit
that referenced
this pull request
Nov 13, 2020
…ed subscription (#8545) ### Motivation 1. Expose consumer names after the mark delete position for the Key_Shared subscription. 2. Remove the consumer from the recenlyJoinedConsumer depends on the valid next position of the next position. Previously, we use the position.nextPosition to decide to remove the consumer from the recenlyJoinedConsumer but this will lead to consumers can't be deleted property. For example, if ledger rollover and the mark delete position is the last position of the old ledger and the max read position is the first position of the new ledger, In this situation, we should remove the consumer from the recenlyJoinedConsumer but in fact, it will stay in the recenlyJoinedConsumer because the max read position always greater than the `markDeletePosition.nextPosition`. So we should get the valid next position for the mark delete position. Related to #8499
codelipenghui
added a commit
that referenced
this pull request
Nov 13, 2020
…ed subscription (#8545) 1. Expose consumer names after the mark delete position for the Key_Shared subscription. 2. Remove the consumer from the recenlyJoinedConsumer depends on the valid next position of the next position. Previously, we use the position.nextPosition to decide to remove the consumer from the recenlyJoinedConsumer but this will lead to consumers can't be deleted property. For example, if ledger rollover and the mark delete position is the last position of the old ledger and the max read position is the first position of the new ledger, In this situation, we should remove the consumer from the recenlyJoinedConsumer but in fact, it will stay in the recenlyJoinedConsumer because the max read position always greater than the `markDeletePosition.nextPosition`. So we should get the valid next position for the mark delete position. Related to #8499 (cherry picked from commit 6867262)
codelipenghui
added a commit
to streamnative/pulsar-archived
that referenced
this pull request
Nov 13, 2020
1. Fix race condition when calling `acknowledgementWasProcessed()`. Currently, when the broker received a message acknowledge request, the broker will call ManagedCursor to delete the position by using `cursor.asyncDelete()`. We should call the acknowledgementWasProcessed() in the delete position callback. Otherwise, when the dispatcher for the Key_Shared subscription will get an unchanged mark delete position so that in some case the consumer can't be removed from the `recentJoinedConsumers`. This might cause the message consumption stuck in the Key_Shared subscription. 2. Currently, the consumer removed from the `recentJoinedConsumers` until the mark delete position is greater than the read position that the consumer joined and there are new messages need to deliver to this consumer. But in some case, there are no new messages need to deliver to this consumer, so this consumer can't be moved from the `recentJoinedConsumers`. This will also lead to consumption stuck. (cherry picked from commit 8df7364) (cherry picked from commit 7cf2d12)
codelipenghui
added a commit
to streamnative/pulsar-archived
that referenced
this pull request
Nov 13, 2020
…ed subscription (apache#8545) 1. Expose consumer names after the mark delete position for the Key_Shared subscription. 2. Remove the consumer from the recenlyJoinedConsumer depends on the valid next position of the next position. Previously, we use the position.nextPosition to decide to remove the consumer from the recenlyJoinedConsumer but this will lead to consumers can't be deleted property. For example, if ledger rollover and the mark delete position is the last position of the old ledger and the max read position is the first position of the new ledger, In this situation, we should remove the consumer from the recenlyJoinedConsumer but in fact, it will stay in the recenlyJoinedConsumer because the max read position always greater than the `markDeletePosition.nextPosition`. So we should get the valid next position for the mark delete position. Related to apache#8499 (cherry picked from commit 6867262) (cherry picked from commit 7b55504)
flowchartsman
pushed a commit
to flowchartsman/pulsar
that referenced
this pull request
Nov 17, 2020
### Motivation 1. Fix race condition when calling `acknowledgementWasProcessed()`. Currently, when the broker received a message acknowledge request, the broker will call ManagedCursor to delete the position by using `cursor.asyncDelete()`. We should call the acknowledgementWasProcessed() in the delete position callback. Otherwise, when the dispatcher for the Key_Shared subscription will get an unchanged mark delete position so that in some case the consumer can't be removed from the `recentJoinedConsumers`. This might cause the message consumption stuck in the Key_Shared subscription. 2. Currently, the consumer removed from the `recentJoinedConsumers` until the mark delete position is greater than the read position that the consumer joined and there are new messages need to deliver to this consumer. But in some case, there are no new messages need to deliver to this consumer, so this consumer can't be moved from the `recentJoinedConsumers`. This will also lead to consumption stuck.
flowchartsman
pushed a commit
to flowchartsman/pulsar
that referenced
this pull request
Nov 17, 2020
…ed subscription (apache#8545) ### Motivation 1. Expose consumer names after the mark delete position for the Key_Shared subscription. 2. Remove the consumer from the recenlyJoinedConsumer depends on the valid next position of the next position. Previously, we use the position.nextPosition to decide to remove the consumer from the recenlyJoinedConsumer but this will lead to consumers can't be deleted property. For example, if ledger rollover and the mark delete position is the last position of the old ledger and the max read position is the first position of the new ledger, In this situation, we should remove the consumer from the recenlyJoinedConsumer but in fact, it will stay in the recenlyJoinedConsumer because the max read position always greater than the `markDeletePosition.nextPosition`. So we should get the valid next position for the mark delete position. Related to apache#8499
codelipenghui
added a commit
to streamnative/pulsar-archived
that referenced
this pull request
Nov 21, 2020
1. Fix race condition when calling `acknowledgementWasProcessed()`. Currently, when the broker received a message acknowledge request, the broker will call ManagedCursor to delete the position by using `cursor.asyncDelete()`. We should call the acknowledgementWasProcessed() in the delete position callback. Otherwise, when the dispatcher for the Key_Shared subscription will get an unchanged mark delete position so that in some case the consumer can't be removed from the `recentJoinedConsumers`. This might cause the message consumption stuck in the Key_Shared subscription. 2. Currently, the consumer removed from the `recentJoinedConsumers` until the mark delete position is greater than the read position that the consumer joined and there are new messages need to deliver to this consumer. But in some case, there are no new messages need to deliver to this consumer, so this consumer can't be moved from the `recentJoinedConsumers`. This will also lead to consumption stuck. (cherry picked from commit 8df7364) (cherry picked from commit 7cf2d12) (cherry picked from commit c0facfd)
codelipenghui
added a commit
to streamnative/pulsar-archived
that referenced
this pull request
Nov 21, 2020
…ed subscription (apache#8545) 1. Expose consumer names after the mark delete position for the Key_Shared subscription. 2. Remove the consumer from the recenlyJoinedConsumer depends on the valid next position of the next position. Previously, we use the position.nextPosition to decide to remove the consumer from the recenlyJoinedConsumer but this will lead to consumers can't be deleted property. For example, if ledger rollover and the mark delete position is the last position of the old ledger and the max read position is the first position of the new ledger, In this situation, we should remove the consumer from the recenlyJoinedConsumer but in fact, it will stay in the recenlyJoinedConsumer because the max read position always greater than the `markDeletePosition.nextPosition`. So we should get the valid next position for the mark delete position. Related to apache#8499 (cherry picked from commit 6867262) (cherry picked from commit 7b55504) (cherry picked from commit e1abbe5)
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
Fix race condition when calling
acknowledgementWasProcessed(). Currently, when the broker received a message acknowledge request, the broker will call ManagedCursor to delete the position by usingcursor.asyncDelete(). We should call the acknowledgementWasProcessed() in the delete position callback. Otherwise, when the dispatcher for the Key_Shared subscription will get an unchanged mark delete position so that in some case the consumer can't be removed from therecentJoinedConsumers. This might cause the message consumption stuck in the Key_Shared subscription.Currently, the consumer removed from the
recentJoinedConsumersuntil the mark delete position is greater than the read position that the consumer joined and there are new messages need to deliver to this consumer. But in some case, there are no new messages need to deliver to this consumer, so this consumer can't be moved from therecentJoinedConsumers. This will also lead to consumption stuck.Verifying this change
All of the CI tests passed.
Does this pull request potentially affect one of the following parts:
If
yeswas chosen, please highlight the changesDocumentation