Introduce ActiveConsumerListener for realizing if a consumer is active in a failover subscription group #1156

sijie · 2018-01-31T02:07:58Z

Motivation

This closes #1058

Modifications

introduce CommandConsumerGroupChange for notifying consumers with consumer group changes
bump the protocol version to v11
introduce ConsumerGroupListener for listening to group changes

Result

The consumer is able to register a ConsumerGroupListener to listen on group changes.

… in a failover subscription group

merlimat · 2018-02-05T18:33:37Z

...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java

+            return false;
+        } else {
+            // If the active consumer is changed, send notification.
+            notifyConsumerGroupChanged(activeConsumer);


The actual change in active consumer can happening a bit later. There is an artificial delay that is in place for a couple of reasons:

Avoid flickering when multiple consumer are disconnecting/reconnecting (eg: during a failover scenario). Ideally we don't want to jump around and elect many active consumers when that happens.

Avoid "most" duplicates due to in-flight (not acked) messages that get resent to the newly elected active consumer.

Here, between this time, when a consumer is elected and when it actually start to get messages, we should avoid notifying that it is "active", because it might get demoted again before it actually gets any messages.

@merlimat I have changed to notify active consumer changes only after the cursor is rewinded. please review it again.

merlimat · 2018-02-05T18:37:23Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java

+     * @see #setConsumerGroupListener(ConsumerGroupListener)
+     * @since 1.22.0
+     */
+    public ConsumerGroupListener getConsumerGroupListener() {


Since we don't use the termConsumerGroup anywhere else, but rather Subscription and Consumer, I think we should be sticking with it. Something like "ActiveConsumerListener"?

Also what would be the semantic on setting the listener on Shared/Exclusive subscriptions?

I renamed "ConsumerGroupListener" to "ActiveConsumerListener".

I added the logic on pulsar client to fail subscribe if a listener is set for non-failover subscription.

…ver subscription.

sijie · 2018-02-10T04:25:15Z

@merlimat I have addressed your comments. can you review this again? After it looks good, I will rebase it to latest master.

merlimat · 2018-02-10T22:25:53Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

+            log.debug("notify consumer {} - that [{}] for subscription {} has new active consumer : {}",
+                consumerId, topicName, subscription.getName(), activeConsumerId);
+        }
+        cnx.ctx().write(Commands.newConsumerGroupChange(consumerId, activeConsumerId));


Use writeAndFlush(), otherwise the message won't be written on the socket until and unrelated flush is triggered.

Also, pass cnx.ctx().voidPromise() as the 2nd parameter to avoid creating the write promise since we don't care here.

merlimat · 2018-02-10T22:28:20Z

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java

    }

+    void notifyConsumerGroupChange(long activeConsumerId) {
+        if (cnx.getRemoteEndpointProtocolVersion() < ProtocolVersion.v11.getNumber()) {


I was thinking here that we could have more descriptive methods in Commands class to group all these feature checks in a single place. Eg:

if (!Commands.peerSupportsActiveConsumerListener(cnx)) { return; } ...

(I know in other places we're checking the version like this, and we should also change separately from this PR)

One thing we need to be careful here is that C++ code gets compiled and automatically will present itself with the latest version v11 (or whatever).

In this case, we should make sure that C++ client can at least discard the notification

merlimat · 2018-02-10T22:41:01Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/ActiveConsumerListener.java

+/**
+ * Listener on the consumer state changes.
+ */
+public interface ActiveConsumerListener {


If we use a single method, it becomes accessible through a lambda. Eg:

interface ActiveConsumerListener { void statusChanged(Consumer, int partitionId, Status status); }

I think lambda is good for non-branchable operations. However it doesn't fit well for branchable operations. we can make a single method here using Status, however the application logic has to write if-else block to take different actions. It is much clear to me if we have two different methods for two different actions. so I chose to have separated methods for handling become active and become inactive, when applications are wring the listener, you just need to think what should do when it become active and what should do when it become inactive.

However if you think using Status and having a lambda interface is much better, let me know. I will make the change.

Sure, that makes sense.

One other thing that I was thinking is that we could have this listener to be more general oriented, so that we could in future reuse it for other kinds of notifications.

Something like ConsumerEventsListener so that is more neutral? What do you think?

merlimat · 2018-02-10T22:41:57Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ConsumerImpl.java

        }
    }

+    void consumerGroupChanged(long activeConsumerId) {


consumerGroupChanged() --> activeConsumerChanged()

merlimat · 2018-02-10T22:44:09Z

pulsar-client/src/main/java/org/apache/pulsar/client/impl/PulsarClientImpl.java

                    new PulsarClientException.InvalidConfigurationException("Consumer configuration undefined"));
        }

+        if (conf.getActiveConsumerListener() != null


I know it doesn't make a lot of sense to check for active/inactive on a non failover subscription. But wouldn't be in a way still semantically correct to always notify of "active" in exclusive/shared mode?

I think the main motivation for this change here is for supporting failover subscription. I would disable this feature for other subscriptions for now, and introduce it if we really need it. does that make sense to you?

merlimat · 2018-02-10T22:44:43Z

pulsar-common/src/main/proto/PulsarApi.proto

 }

+// changes on consumer group
+message CommandConsumerGroupChange {


CommandActiveConsumerChanged

merlimat · 2018-02-10T22:49:07Z

...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java

-    protected void pickAndScheduleActiveConsumer() {
+    protected void notifyConsumerGroupChanged(Consumer activeConsumer) {
+        if (null != activeConsumer && subscriptionType == SubType.Failover) {
+            consumers.forEach(consumer ->


We should try to avoid to send multiple "inactive" notifications to a consumer.
If I was already "inactive", I don't need to be notified when a new consumer is promoted and I'm not involved.

I think this is the first iteration for this listener. I am trying to be avoiding any optimization at the beginning and deliver notifications for every changes. This way we can make sure the first version is "eventually" correct without hitting any race condition on handling this kind of smart logic. If sending multiple notifications become a problem we can do a subsequent change later.

does that make sense to you?

merlimat · 2018-02-10T22:53:34Z

...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java

+    protected void notifyConsumerGroupChanged(Consumer activeConsumer) {
+        if (null != activeConsumer && subscriptionType == SubType.Failover) {
+            consumers.forEach(consumer ->
+                consumer.notifyConsumerGroupChange(activeConsumer.consumerId()));


The consumerId is a per-connection identifier (unique within a single TCP connection), but different connections from different clients can very well use the same Id. (The id is just a shortcut to refer to a particular topic/subscription without passing the names each time).

For consumers on a single subscription, there is no single unique identifier. consumerName is generally a random generated string, though there is collision detection (since collisions wont affect the correctness) and it can be manually overridden.

I think the only way is to use the Consumer object identify.

merlimat · 2018-02-10T22:54:37Z

...r/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java

+            // the active consumer is not changed
+            Consumer currentActiveConsumer = ACTIVE_CONSUMER_UPDATER.get(this);
+            if (null == currentActiveConsumer) {
+                log.warn("Current active consumer disappears while adding consumer {}", consumer);


This should be a debug log

sijie · 2018-02-13T16:40:25Z

@merlimat I have addressed your comments, please take another round of review. If this looks good for you, I will rebase it to latest master.

merlimat · 2018-02-14T06:25:57Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerConfiguration.java

+    /**
+     * @return this configured {@link ActiveConsumerListener} for the consumer.
+     * @see #setActiveConsumerListener(ActiveConsumerListener)
+     * @since 1.22.0


merlimat · 2018-02-14T06:29:55Z

@sijie Change looks good, though there's still the issue at https://github.com/apache/incubator-pulsar/pull/1156/files#r167413443 . I think is using the consumerId of the active consumer to decide if other consumers are active/inactive, but we cannot compare consumerId across consumers.

The other comment was around making the Listener interface name more neutral so that we can add more event handlers, if (and when) the need arises, without having to add an additional listener.

Use consumer object as identifier for comparison

sijie · 2018-02-14T15:58:25Z

@merlimat as you suggested, I did two changes here:

rename ActiveConsumerListener to ConsumerEventListener. currently since we only use it for failover subscription, so I am still disallowing shared/exclusive subscription to use the listener. we can enable if we support more events in future.
I changed the dispatcher to use consumer object as the comparison identifier.

merlimat

👍

ivankelly

Does this provide a way to figure out the current active consumer?

This could also have been done by each consumer periodically adding a message to the topic. if you start getting the messages, you are the active consumer.

ivankelly · 2018-02-14T16:04:34Z

pulsar-client/src/main/java/org/apache/pulsar/client/api/ConsumerEventListener.java

+    /**
+     * Notified when the consumer group is changed, and the consumer becomes the active consumer.
+     */
+    void becomeActive(Consumer consumer, int partitionId);


merlimat · 2018-02-14T16:23:30Z

Does this provide a way to figure out the current active consumer?

Not right now, I think it would be interesting to mention who's the active consumer in the topic in the topics stats. We could have a Boolean field in the ConsumerStats that is set to null in non-failover subscriptions.

sijie · 2018-02-14T16:40:01Z

This could also have been done by each consumer periodically adding a message to the topic. if you start getting the messages, you are the active consumer.

@ivankelly : yes that's the trick we are using now for the other project.

…leader

sijie · 2018-02-14T17:46:09Z

rebased to the latest master

sijie · 2018-02-14T23:31:45Z

the CI failed with license header. fixed license header.

I forgot another item to make CPP client ignore this notification.

…leader

merlimat

👍

Introduce ConsumerGroupListener for realizing if a consumer is active…

bb30631

… in a failover subscription group

sijie added this to the 2.0.0-incubating milestone Jan 31, 2018

sijie self-assigned this Jan 31, 2018

sijie requested review from merlimat and rdhabalia January 31, 2018 02:07

merlimat added the type/feature The PR added a new feature or issue requested a new feature label Jan 31, 2018

merlimat reviewed Feb 5, 2018

View reviewed changes

sijie added 3 commits February 10, 2018 11:30

Rename ConsumerGroupListener to ActiveConsumerListener

d4f2820

Fail subscribe if active consumer listener is provided with non-failo…

e8399af

…ver subscription.

Notify consumer state change only after the cursor is rewinded.

44adf2f

merlimat changed the title ~~Introduce ConsumerGroupListener for realizing if a consumer is active in a failover subscription group~~ Introduce ActiveConsumerListener for realizing if a consumer is active in a failover subscription group Feb 10, 2018

merlimat reviewed Feb 10, 2018

View reviewed changes

Address comments

0317289

merlimat reviewed Feb 14, 2018

View reviewed changes

Rename ActiveConsumerListener to ConsumerEventListener

b8411ba

Use consumer object as identifier for comparison

merlimat approved these changes Feb 14, 2018

View reviewed changes

ivankelly approved these changes Feb 14, 2018

View reviewed changes

sijie added 2 commits February 15, 2018 01:18

Merge remote-tracking branch 'apache/master' into notification_whois_…

ea6f5be

…leader

Fix test after rebase

91fe3a2

Fix license headers

28559f1

sijie mentioned this pull request Feb 14, 2018

Implement ActiveConsumerListener for pulsar cpp client #1240

Closed

Ignore active consumer change command in cpp client

358c772

Merge remote-tracking branch 'apache/master' into notification_whois_…

0dc1638

…leader

merlimat approved these changes Feb 15, 2018

View reviewed changes

rename "become" to "became"

647624b

merlimat merged commit f2160c0 into apache:master Feb 15, 2018

sijie mentioned this pull request Apr 25, 2018

Improve failover subscription for non-partitioned topic #1267

Closed

rdhabalia mentioned this pull request Jul 10, 2018

Delayed acks impl #1462

Merged

sijie mentioned this pull request Dec 27, 2019

ISSUE-1240: Implement ActiveConsumerListener for pulsar cpp client streamnative/pulsar-archived#467

Closed

Introduce ActiveConsumerListener for realizing if a consumer is active in a failover subscription group #1156

Introduce ActiveConsumerListener for realizing if a consumer is active in a failover subscription group #1156

Uh oh!

Conversation

sijie commented Jan 31, 2018

Motivation

Modifications

Result

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sijie commented Feb 10, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

sijie commented Feb 13, 2018

Uh oh!

Choose a reason for hiding this comment

Uh oh!

merlimat commented Feb 14, 2018

Uh oh!

sijie commented Feb 14, 2018

Uh oh!

merlimat left a comment

Choose a reason for hiding this comment

Uh oh!

ivankelly left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

merlimat commented Feb 14, 2018

Uh oh!

sijie commented Feb 14, 2018

Uh oh!

sijie commented Feb 14, 2018

Uh oh!

sijie commented Feb 14, 2018

Uh oh!

merlimat left a comment

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development