Able to handling messages with multiple listener threads in order for the Key_Shared subscription. #9329

codelipenghui · 2021-01-26T16:43:43Z

Motivation

Currently, a consumer is pinged to a given listener thread to ensure the message is processed ordered for a topic. But for the Key_Shared subscription, the message process order is based on the message key, not the order of the topic. So this PR is able to handle messages with multiple listener threads for the Key_Shared subscription which also keeps the order of message key.

Verifying this change

New tests added.

Does this pull request potentially affect one of the following parts:

If yes was chosen, please highlight the changes

Dependencies (does it add or upgrade a dependency): (no)
The public API: (no)
The schema: (no
The default values of configurations: (no)
The wire protocol: (no)
The rest endpoints: (no)
The admin cli options: (no)
Anything that affects deployment: (no)

Documentation

Does this pull request introduce a new feature? (no)

codelipenghui · 2021-01-27T00:07:32Z

/pulsarbot run-failure-checks

hangc0276 · 2021-01-27T02:30:14Z

/pulsarbot run-failure-checks

codelipenghui · 2021-01-27T05:22:18Z

/pulsarbot run-failure-checks

sijie

Overall the change looks good to me. Just one comment about the OrderedExecutor.

sijie · 2021-01-27T06:11:15Z

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

@@ -80,6 +83,7 @@
    protected volatile long incomingMessagesSize = 0;
    protected volatile Timeout batchReceiveTimeout = null;
    protected final Lock reentrantLock = new ReentrantLock();
+    private OrderedExecutor orderedExecutor = null;


Can you just change the listenerExecutor to OrderedExecutor?

Good suggestion!

Oh @sijie, I have checked the details seems currently the listenerExecutor used by some other operations and the listener executor is a ScheduledExecutorService, the consumer will always get a listenerExecutor with 1 thread, so we can't change the listenerExecutor to OrderedExecutor directly.

How about we use OrderedScheduler? BookKeeper common library also provides an OrderedScheduler.

I have addressed the comment, please take a look.

…the Key_Shared subscription.

…listener_ordered

codelipenghui · 2021-02-05T00:38:59Z

/pulsarbot run-failure-checks

…listener_ordered

sijie · 2021-02-11T03:45:20Z

/pulsarbot run-failure-checks

sijie · 2021-02-12T01:54:16Z

@codelipenghui I think there are shading issues. Can you fix it?

codelipenghui · 2021-02-16T13:10:33Z

@sijie Yes, the interesting thing is I follow the steps of the github workflow, it can get passed on my laptop. And I have verified the pulsar-client and pulsar-client-all locally, also works.

codelipenghui · 2021-02-17T02:44:03Z

@merlimat Sorry, I missed the community meeting. I have updated the description of this PR. This PR is not to try to fix the ordering issue of the consumer under the Key_Shared subscription, this PR is to try to allow a consumer to use multiple listener threads for the Key_Shared subscription and keep the message with the same key processed by the same thread.

…listener_ordered

merlimat · 2021-02-18T00:20:32Z

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

@@ -69,7 +70,8 @@
    protected final CompletableFuture<Consumer<T>> subscribeFuture;
    protected final MessageListener<T> listener;
    protected final ConsumerEventListener consumerEventListener;
-    protected final ExecutorService listenerExecutor;
+    protected final ExecutorProvider executorProvider;
+    protected final ScheduledExecutorService pingedExecutor;


It's not clear to me what pingedExecutor means here. Does that mean "pinned"?

Oh, sorry. pinned

merlimat · 2021-02-18T05:38:44Z

pulsar-client/src/main/java/org/apache/pulsar/client/util/ExecutorProvider.java

+        return getExecutor(object == null ? -1 : object.hashCode() & Integer.MAX_VALUE);
+    }
+
+    public ExecutorService getExecutor(int hash) {


In general, from API perspective, it would be better to just expose an Object (or String) for selecting the thread. Couldn't we calculate the hash internally always?

Ok, we can move the key hash calculate internally and replace this method to getExecutor(byte[]).

codelipenghui · 2021-02-18T09:31:40Z

@merlimat I have addressed the comments, PTAL.

- apache#9630 and apache#9329 were merged concurrently which caused the master branch to break - fixes SpotBugs error "Medium: org.apache.pulsar.client.impl.ConsumerBase.NONE_KEY should be package protected [org.apache.pulsar.client.impl.ConsumerBase] At ConsumerBase.java:[line 887] MS_PKGPROTECT"

… the Key_Shared subscription. (apache#9329) ### Motivation Currently, a consumer is pinged to a given listener thread to ensure the message is processed ordered for a topic. But for the Key_Shared subscription, the message process order is based on the message key, not the order of the topic. So this PR is able to handle messages with multiple listener threads for the Key_Shared subscription which also keeps the order of message key. (cherry picked from commit e4523e0)

… the Key_Shared subscription. (#9329) Currently, a consumer is pinged to a given listener thread to ensure the message is processed ordered for a topic. But for the Key_Shared subscription, the message process order is based on the message key, not the order of the topic. So this PR is able to handle messages with multiple listener threads for the Key_Shared subscription which also keeps the order of message key. (cherry picked from commit e4523e0)

codelipenghui requested review from merlimat, sijie, 315157973, hangc0276, jiazhai and zymap January 26, 2021 16:44

codelipenghui self-assigned this Jan 26, 2021

codelipenghui added the type/enhancement The enhancements for the existing features or docs. e.g. reduce memory usage of the delayed messages label Jan 26, 2021

codelipenghui added this to the 2.8.0 milestone Jan 26, 2021

codelipenghui added the release/2.7.1 label Jan 26, 2021

sijie reviewed Jan 27, 2021

View reviewed changes

Able to handling messages with multiple listener thread in order for …

7a418f3

…the Key_Shared subscription.

codelipenghui force-pushed the penghui/key_shared_listener_ordered branch from 63e7271 to 7a418f3 Compare January 28, 2021 13:23

codelipenghui added 3 commits February 5, 2021 00:04

Merge remote-tracking branch 'apache/master' into penghui/key_shared_…

222ccac

…listener_ordered

Apply comments.

22998dd

Apply comments.

fb86cc0

sijie approved these changes Feb 5, 2021

View reviewed changes

zymap approved these changes Feb 5, 2021

View reviewed changes

315157973 approved these changes Feb 5, 2021

View reviewed changes

codelipenghui added 6 commits February 5, 2021 21:36

Merge remote-tracking branch 'apache/master' into penghui/key_shared_…

b7dffba

…listener_ordered

Fix shaded.

cb02fba

Merge remote-tracking branch 'apache/master' into penghui/key_shared_…

28fbf11

…listener_ordered

Fix test

4f449c9

Fix test

bf21269

Fix shaded

4b25f93

Merge branch 'master' into penghui/key_shared_listener_ordered

a85a4da

Add bookkeeper-common dependency to pulsar-client.

7240086

codelipenghui added 2 commits February 17, 2021 14:50

Remove bookkeeper-common dependency from pulsar-client.

bc32f6e

Merge remote-tracking branch 'apache/master' into penghui/key_shared_…

a2ce3b2

…listener_ordered

merlimat reviewed Feb 18, 2021

View reviewed changes

Apply comments.

368485b

codelipenghui merged commit e4523e0 into apache:master Feb 21, 2021

codelipenghui deleted the penghui/key_shared_listener_ordered branch February 21, 2021 03:12

zymap added the cherry-picked/branch-2.7 Archived: 2.7 is end of life label Mar 1, 2021

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Able to handling messages with multiple listener threads in order for the Key_Shared subscription. #9329

Able to handling messages with multiple listener threads in order for the Key_Shared subscription. #9329

codelipenghui commented Jan 26, 2021 •

edited

Loading

codelipenghui commented Jan 27, 2021

hangc0276 commented Jan 27, 2021

codelipenghui commented Jan 27, 2021

sijie left a comment

sijie Jan 27, 2021

codelipenghui Jan 27, 2021

codelipenghui Jan 28, 2021

sijie Feb 1, 2021

codelipenghui Feb 4, 2021

codelipenghui commented Feb 5, 2021

sijie commented Feb 11, 2021

sijie commented Feb 12, 2021

codelipenghui commented Feb 16, 2021

codelipenghui commented Feb 17, 2021

merlimat Feb 18, 2021

codelipenghui Feb 18, 2021

merlimat Feb 18, 2021

codelipenghui Feb 18, 2021

codelipenghui commented Feb 18, 2021

Able to handling messages with multiple listener threads in order for the Key_Shared subscription. #9329

Able to handling messages with multiple listener threads in order for the Key_Shared subscription. #9329

Conversation

codelipenghui commented Jan 26, 2021 • edited Loading

Motivation

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

codelipenghui commented Jan 27, 2021

hangc0276 commented Jan 27, 2021

codelipenghui commented Jan 27, 2021

sijie left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Feb 5, 2021

sijie commented Feb 11, 2021

sijie commented Feb 12, 2021

codelipenghui commented Feb 16, 2021

codelipenghui commented Feb 17, 2021

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codelipenghui commented Feb 18, 2021

codelipenghui commented Jan 26, 2021 •

edited

Loading