[C++] Support setting priority for the consumer #12076
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
merlimat
approved these changes
Sep 17, 2021
BewareMyPower
requested changes
Sep 18, 2021
BewareMyPower
approved these changes
Sep 18, 2021
|
There's something wrong with vcpkg build on Windows currently so that the I'll push a fix soon. |
|
Please rebase to master to fix the broken CI. |
metahys
force-pushed
the
cpp-client-priority
branch
from
ac383c6
to
cb4a41e
Compare
PTAL @BewareMyPower, running the workflows require approval. |
|
It seems that |
Yeah, it's a flaky test but now it doesn't fail frequently. You can comment |
OK, got it. Thanks. |
4 tasks
BewareMyPower
added a commit
that referenced
this pull request
Jan 27, 2022
…13883) In C++ client, there is a corner case that when a reader's start message ID is the last message of a topic, `hasMessageAvailable` returns true. However, it should return false because the start message ID is exclusive and in this case `readNext` would never return a message unless new messages arrived. The current C++ implementation of `hasMessageAvailable` is from long days ago and has many problems. So this PR migrates the Java implementation of `hasMessageAvailable` to C++ client. Since after the modifications we need to access `startMessageId` in `hasMessageAvailable`, which is called in a different thread from `connectionOpened` that might modify `startMessageId`. We use a common mutex `mutexForMessageIds` to protect the access to `lastDequedMessageId_` and `lastMessageIdInBroker_`. To fix the original tests when `startMessageId` is latest, this PR adds a `GetLastMessageIdResponse` as the response of `GetLastMessageId` request. The `GetLastMessageIdResponse` contains the `consumer_mark_delete_position` introduced from #9652 to compare with `last_message_id` when `startMessageId` is latest. This change added tests `ReaderTest#testHasMessageAvailableWhenCreated` and `MessageIdTest# testCompareLedgerAndEntryId`. (cherry picked from commit e50493e) Fix the conflicts by: - Remove ReaderImpl::getLastMessageIdAsync introduced from #11723 - Remove getPriorityLevel() method introduced from #12076 - Revert changes of registerConsumer from #12118 - Remove new fields introduced from #13627
bharanic-dev
pushed a commit
to bharanic-dev/pulsar
that referenced
this pull request
Mar 18, 2022
* [C++] Support setting priority for the consumer * Fixed formatting * Fixed code style Co-authored-by: sunhongyi <sunhongyi@tencent.com>
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
#165 added support for dispatching message based on consumer priority. However, for C++ client, there is no way to set such priority on
ConsumerConfiguration. This PR is to support setting priority for the consumer in C++ client.Modifications
getPriorityLevelandsetPriorityLevelmethods inConsumerConfigurationCommands::newSubscribeto add additional parameterpriority_levelwith default value 0Verifying this change
This change added tests and can be verified as follows:
ConsumerConfiguration::getPriorityLevelandConsumerConfiguration::setPriorityLevelinConsumerConfigurationTestDoes this pull request potentially affect one of the following parts:
If
yeswas chosen, please highlight the changes