[Issue 1240][C++] Support setting ConsumerEventListener in pulsar cpp client #12118
Conversation
This file contains hidden or 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
Contributor
Author
|
/pulsarbot run-failure-checks |
merlimat
reviewed
Sep 23, 2021
|
|
||
| class DummyEventListener : public ConsumerEventListener { | ||
| public: | ||
| virtual void becomeActive(Consumer consumer, int partitionId) override {} |
Contributor
There was a problem hiding this comment.
We should add a test that verifies that the active/inactive tests are being invoked. We can follow the same line of test that we're doing for java
Contributor
Author
There was a problem hiding this comment.
I have added unit tests in ConsumerTest. And I renamed "become" to "became" to make it consistent with java client. The call to registerConsumer is moved from handleCreateConsumer to connectionOpened, so that we can handle commands such as ACTIVE_CONSUMER_CHANGE properly.
@merlimat Please take a look.
merlimat
added
the
type/enhancement
merlimat
approved these changes
Sep 24, 2021
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
… client (apache#12118) * [C++] Support setting ConsumerEventListener in pulsar cpp client * Rename "become" to "became" * Register consumer before sending subscribe request * Add unit tests in ConsumerTest
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
5 participants
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.
Fixes #1240
Motivation
This PR supports setting
ConsumerEventListenerin pulsar cpp client.Modifications
ConsumerEventListenerfor listening consumer events.setConsumerEventListenerandgetConsumerEventListenermethods inConsumerConfigurationhandleActiveConsumerChangemethod inClientConnectionto handle theACTIVE_CONSUMER_CHANGEevent.Verifying this change
ConsumerConfigurationTestConsumerTestDoes this pull request potentially affect one of the following parts:
Documentation