Fix hasMessageAvailable might return true after seeking to latest #409
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
BewareMyPower
changed the title
BewareMyPower
force-pushed
the
bewaremypower/has-msg-available-after-seek-latest
branch
2 times, most recently
from
f59e148
to
886ff1c
Compare
After a seek operation is done, the `startMessageId` will be updated until the reconnection due to the seek is done in `connectionOpened`. So before it's updated, `hasMessageAvailable` could compare with an outdated `startMessageId` and return a wrong value. ### Modifications Replace `duringSeek` with a `SeekStatus` field: - `NOT_STARTED`: initial, or a seek operation is done. `seek` could only succeed in this status. - `IN_PROGRESS`: A seek operation has started but the client does not receive the response from broker. - `COMPLETED`: The client has received the seek response but the seek future is not done. After the status becomes `COMPLETED`, if the connection is not ready, next time the connection is established, the status will change from `COMPLETED` to `NOT_STARTED` and then seek future will be completed in the internal executor. Add `testHasMessageAvailableAfterSeekToEnd` and `testSeekInProgress`.
BewareMyPower
changed the title
BewareMyPower
force-pushed
the
bewaremypower/has-msg-available-after-seek-latest
branch
from
886ff1c
to
0e91bbd
Compare
BewareMyPower
changed the title
BewareMyPower
requested review from
RobertIndie,
shibd,
Demogorgon314 and
merlimat
|
@merlimat @shibd @RobertIndie @Demogorgon314 Could you take a look? |
shibd
reviewed
Mar 11, 2024
RobertIndie
reviewed
Mar 11, 2024
lib/ConsumerImpl.cc
Outdated
| @@ -236,16 +236,15 @@ Future<Result, bool> ConsumerImpl::connectionOpened(const ClientConnectionPtr& c | |||
| // sending the subscribe request. | |||
| cnx->registerConsumer(consumerId_, get_shared_this_ptr()); | |||
|
|
|||
| if (duringSeek_) { | |||
| if (duringSeek()) { | |||
| ackGroupingTrackerPtr_->flushAndClean(); | |||
| } | |||
|
|
|||
| Lock lockForMessageId(mutexForMessageId_); | |||
| // Update startMessageId so that we can discard messages after delivery restarts | |||
There was a problem hiding this comment.
We can move this comment inside the clearReceiveQueue.
|
@RobertIndie @shibd Comments are addressed, PTAL again. |
Demogorgon314
approved these changes
Mar 11, 2024
RobertIndie
approved these changes
Mar 11, 2024
shibd
approved these changes
Mar 11, 2024
BewareMyPower
deleted the
bewaremypower/has-msg-available-after-seek-latest
branch
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
See apache/pulsar-client-python#199
There is a race condition when
hasMessageAvailableis called afterseekif the start message ID ofReaderis earliest.In
ConsumerImpl::hasMessageAvailableAsync, if the connection is not established at the moment,lastDequedMessageId_will beearliestbecause no message is received. SincelastMessageIdInBroker_is alsoearliest,getLastMessageIdAsyncwill be called and then it comes atpulsar-client-cpp/lib/ConsumerImpl.cc
Line 1554 in e2cacb7
However, before
getLastMessageIdAsyncis called,messageIdwasearliestbecauselastDequedMessageId_andstartMessageId_were bothearliest. However, when the callback is called, thestartMessageId_has already been updated tolatestinconnectionOpened, so we should compare tolatest.Modifications
In the callback of
getLastMessageIdAsync, retrieve the latest value ofstartMessageId_to compare rather then reusing the old value.Refactor the seek flow to reset the seek states and trigger the callback after updating the
startMessageId_.ReaderTest.testHasMessageAvailableAfterSeekToEndis added to cover the changes.