From 521e5763be96b60d698d4674f3ed8e5bc7e0c143 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Thu, 13 Jun 2024 19:27:09 +0800 Subject: [PATCH 01/10] [improve][pip] PIP-359: Support custom message listener executor for specific subscription --- pip/pip-359.md | 221 +++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 221 insertions(+) create mode 100644 pip/pip-359.md diff --git a/pip/pip-359.md b/pip/pip-359.md new file mode 100644 index 0000000000000..5f7de718ea0f9 --- /dev/null +++ b/pip/pip-359.md @@ -0,0 +1,221 @@ + + +# PIP-359: Support custom message listener executor for specific subscription + + + + +Implementation PR: [#22861](https://github.com/apache/pulsar/pull/22861) +# Motivation + + + +In the current Pulsar client versions, all consumer instances created from the same PulsarClient +will share a thread pool(`ExecutorProvider`) to process message listeners, and sometimes there are multiple +consumer message listeners bound to the same thread. Obviously, when a consumer +processes messages and causes long-term blocking, it will cause the messages of +other consumers bound to the same thread to fail to be processed in time, resulting in +message delays. Therefore, for this scenario, it may be necessary to support specific a message +listener executor of consumer latitudes to avoid mutual influence between different consumers. +# Goals +1. Improve consumer message listener isolation, solve the consumption delay problem caused by +mutual influence of different consumers from the same PulsarClient instance. + +## In Scope + +Nothing. + +## Out of Scope + + +Nothing. + + +# High Level Design +Nothing. + + + +# Detailed Design + +## Design & Implementation Details + + +Add an interface `MessageListenerExecutor`, responsible for executing message listener callback tasks. +Users can customize the implementation to determine in which thread the message listener task is executed. +For example, in the situation described in [Motivation](#motivation) part, users can implement the +interface with an independent underlying thread pool to ensure that the message listener task of each +consumer is executed in a separate thread. The caller would be responsible for the life cycle of the +Executor, and it would be used only for this specific consumer. +```java +public interface MessageListenerExecutor { + + /** + * select a thread by message(if necessary, for example, + * Key_Shared SubscriptionType, maybe need select thread + * by message order key to ensure order) to execute the runnable! + * + * @param message the message + * @param runnable the runnable to execute + */ + void execute(Message message, Runnable runnable); +} +``` + +### Usage Example +```java + private void startConsumerWithMessageListener(String topic, String subscriptionName) { + ConsumerBuilder consumerBuilder = + pulsarClient.newConsumer(Schema.INT64) + .topic(topic) + .subscriptionName(subscriptionName) + .messageListener((c1, msg) -> { + log.info("Received message [{}] in the listener", msg.getValue()); + c1.acknowledgeAsync(msg); + }); + + ExecutorService executor = Executors.newSingleThreadExecutor( + new ExecutorProvider.ExtendedThreadFactory(subscriptionName + "listener-executor-", true)); + // set and then message lister will be executed in the executor + consumerBuilder.messageListenerExecutor((message, runnable) -> executor.execute(runnable)); + + Consumer consumer = consumerBuilder.subscribe(); + } +``` +## Public-facing Changes + + + +### Public API + + + +1. Add an optional config `messageListenerExecutor` in `ConsumerBuilder` +```java +ConsumerBuilder messageListenerExecutor(MessageListenerExecutor messageListenerExecutor); +``` + +### Binary protocol +Nothing. + +### Configuration +Nothing. + +### CLI +Nothing. + +### Metrics + + +Nothing. + +# Monitoring + + +Nothing. + +# Security Considerations + +Nothing. + +# Backward & Forward Compatibility +You can do upgrading or reverting normally, no specified steps are needed to do. +# Alternatives + + +Nothing. + +# General Notes + +# Links + + +* Mailing List discussion thread: +* Mailing List voting thread: From 40b0f3d65a7085c6c7e1d165e65e7f2c2c96dc83 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Fri, 14 Jun 2024 10:38:19 +0800 Subject: [PATCH 02/10] add discussion email link --- pip/pip-359.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 5f7de718ea0f9..3d3852bf2581c 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -217,5 +217,5 @@ Nothing. -* Mailing List discussion thread: +* Mailing List discussion thread: https://lists.apache.org/thread/8nhqfdhkglsg5bgx6z7c1nho7z7l596l * Mailing List voting thread: From 3704b427c1d766a6d8000be658b7c5972588155f Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Mon, 17 Jun 2024 19:53:49 +0800 Subject: [PATCH 03/10] add interface implementation suggestions --- pip/pip-359.md | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/pip/pip-359.md b/pip/pip-359.md index 3d3852bf2581c..b5615c62dee8e 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -116,6 +116,12 @@ public interface MessageListenerExecutor { } ``` +### Interface implementation suggestions +When implementing the `MessageListenerExecutor` interface, you should consider the following points. +1. if you need to ensure the order of message processing, +you can select the thread by the message order key or `msg.getTopicName()`(partition topic name), +to ensure that the messages of the same order key (or partition) are processed in same thread. + ### Usage Example ```java private void startConsumerWithMessageListener(String topic, String subscriptionName) { From 10dd8226fc5046b5964c6ffe60f251eb993406a0 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Tue, 18 Jun 2024 18:26:13 +0800 Subject: [PATCH 04/10] add Background knowledge and refactor API --- pip/pip-359.md | 208 ++++++++++++++----------------------------------- 1 file changed, 58 insertions(+), 150 deletions(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index b5615c62dee8e..1588f74c45daa 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -1,99 +1,76 @@ - - # PIP-359: Support custom message listener executor for specific subscription +Implementation PR: [#22861](https://github.com/apache/pulsar/pull/22861) - +In the current Pulsar client versions, when using a Pulsar Consumer, we have two options to consume messages: +1. Synchronously, by calling `consumer.recieve()` +```java +public class ConsumerExample { + public static void main(String[] args) throws PulsarClientException { + PulsarClient pulsarClient = PulsarClient.builder() + .serviceUrl("pulsar://localhost:6650") + .build(); + Consumer consumer = pulsarClient.newConsumer(Schema.INT64) + .topic("persistent://public/default/my-topic") + .subscriptionName("my-subscription") + .subscribe(); + do { + Message message = consumer.receive(); + consumer.acknowledge(message); + } while (true); - -Implementation PR: [#22861](https://github.com/apache/pulsar/pull/22861) # Motivation - - -In the current Pulsar client versions, all consumer instances created from the same PulsarClient -will share a thread pool(`ExecutorProvider`) to process message listeners, and sometimes there are multiple -consumer message listeners bound to the same thread. Obviously, when a consumer -processes messages and causes long-term blocking, it will cause the messages of -other consumers bound to the same thread to fail to be processed in time, resulting in -message delays. Therefore, for this scenario, it may be necessary to support specific a message -listener executor of consumer latitudes to avoid mutual influence between different consumers. +As [Background knowledge](#background-knowledge) mentioned, when using asynchronous consumer, +by registering a MessageListener interface, there is a problem of different consumer groups +affecting each other, leading to unnecessary consumption delays. +Therefore, for this scenario, this PIP prepare to support specific a message +listener executor of consumer latitudes to solve such problem. # Goals 1. Improve consumer message listener isolation, solve the consumption delay problem caused by mutual influence of different consumers from the same PulsarClient instance. ## In Scope - -Nothing. +If this PIP is accepted, it will help Pulsar solve the problem of different consumers +from same `PulsarClient` affecting each other in the asynchronous consumption mode(`MessageListener`). ## Out of Scope +This PIP will not build the plugin library mentioned in [PR](https://github.com/apache/pulsar/pull/22902#issuecomment-2169962642), +we will open a new PIP in the future to do this - -Nothing. - - -# High Level Design -Nothing. - - # Detailed Design ## Design & Implementation Details - Add an interface `MessageListenerExecutor`, responsible for executing message listener callback tasks. Users can customize the implementation to determine in which thread the message listener task is executed. @@ -125,103 +102,34 @@ to ensure that the messages of the same order key (or partition) are processed i ### Usage Example ```java private void startConsumerWithMessageListener(String topic, String subscriptionName) { + ExecutorService executor = Executors.newSingleThreadExecutor( + new ExecutorProvider.ExtendedThreadFactory(subscriptionName + "listener-executor-", true)); ConsumerBuilder consumerBuilder = pulsarClient.newConsumer(Schema.INT64) .topic(topic) .subscriptionName(subscriptionName) + // set and then message lister will be executed in the executor .messageListener((c1, msg) -> { log.info("Received message [{}] in the listener", msg.getValue()); c1.acknowledgeAsync(msg); - }); - - ExecutorService executor = Executors.newSingleThreadExecutor( - new ExecutorProvider.ExtendedThreadFactory(subscriptionName + "listener-executor-", true)); - // set and then message lister will be executed in the executor - consumerBuilder.messageListenerExecutor((message, runnable) -> executor.execute(runnable)); + }, (message, runnable) -> executor.execute(runnable)); Consumer consumer = consumerBuilder.subscribe(); } ``` ## Public-facing Changes - - ### Public API - - -1. Add an optional config `messageListenerExecutor` in `ConsumerBuilder` +1. Add the following optional overloaded function `messageListener` in `ConsumerBuilder`, to support specifying the messageListener executor ```java -ConsumerBuilder messageListenerExecutor(MessageListenerExecutor messageListenerExecutor); +ConsumerBuilder messageListener(@NonNull MessageListener messageListener, @NonNull MessageListenerExecutor messageListenerExecutor); ``` -### Binary protocol -Nothing. - -### Configuration -Nothing. - -### CLI -Nothing. - -### Metrics - - -Nothing. - -# Monitoring - - -Nothing. - -# Security Considerations - -Nothing. - # Backward & Forward Compatibility You can do upgrading or reverting normally, no specified steps are needed to do. -# Alternatives - - -Nothing. - -# General Notes # Links - * Mailing List discussion thread: https://lists.apache.org/thread/8nhqfdhkglsg5bgx6z7c1nho7z7l596l * Mailing List voting thread: From a6a36311838ce54b578d33af58909b0ea26412fe Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Tue, 18 Jun 2024 21:18:30 +0800 Subject: [PATCH 05/10] update example --- pip/pip-359.md | 89 +++++++++++++++++++++++++++++++++++++++++--------- 1 file changed, 74 insertions(+), 15 deletions(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 1588f74c45daa..3b25c6dcd3ce1 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -101,21 +101,80 @@ to ensure that the messages of the same order key (or partition) are processed i ### Usage Example ```java - private void startConsumerWithMessageListener(String topic, String subscriptionName) { - ExecutorService executor = Executors.newSingleThreadExecutor( - new ExecutorProvider.ExtendedThreadFactory(subscriptionName + "listener-executor-", true)); - ConsumerBuilder consumerBuilder = - pulsarClient.newConsumer(Schema.INT64) - .topic(topic) - .subscriptionName(subscriptionName) - // set and then message lister will be executed in the executor - .messageListener((c1, msg) -> { - log.info("Received message [{}] in the listener", msg.getValue()); - c1.acknowledgeAsync(msg); - }, (message, runnable) -> executor.execute(runnable)); - - Consumer consumer = consumerBuilder.subscribe(); - } + private void startConsumerWithMessageListener(String topic, String subscriptionName) throws PulsarClientException { + // for example: key_shared + MessageListenerExecutor keySharedExecutor = getKeySharedMessageListenerExecutor(subscriptionName); + Consumer keySharedconsumer = + pulsarClient.newConsumer(Schema.INT64) + .topic(topic) + .subscriptionName(subscriptionName) + // set and then message lister will be executed in the executor + .messageListener((c1, msg) -> { + log.info("Received message [{}] in the listener", msg.getValue()); + c1.acknowledgeAsync(msg); + }, keySharedExecutor) + .subscribe(); + + + // for example: partition_ordered + MessageListenerExecutor partitionOrderedExecutor = getPartitionOrderdMessageListenerExecutor(subscriptionName); + Consumer partitionOrderedConsumer = + pulsarClient.newConsumer(Schema.INT64) + .topic(topic) + .subscriptionName(subscriptionName) + // set and then message lister will be executed in the executor + .messageListener((c1, msg) -> { + log.info("Received message [{}] in the listener", msg.getValue()); + c1.acknowledgeAsync(msg); + }, partitionOrderedExecutor) + .subscribe(); + + // for example: out-of-order + ExecutorService executorService = Executors.newFixedThreadPool(10); + Consumer outOfOrderConsumer = + pulsarClient.newConsumer(Schema.INT64) + .topic(topic) + .subscriptionName(subscriptionName) + // not set and then message lister will be executed in the default executor + .messageListener((c1, msg) -> { + log.info("Received message [{}] in the listener", msg.getValue()); + c1.acknowledgeAsync(msg); + }, (message, runnable) -> executorService.execute(runnable)) + .subscribe(); +} + +private static MessageListenerExecutor getKeySharedMessageListenerExecutor(String subscriptionName) { + ExecutorProvider executorProvider = new ExecutorProvider(10, subscriptionName + "listener-executor-"); + + return (message, runnable) -> { + byte[] key = "".getBytes(StandardCharsets.UTF_8); + if (message.hasKey()) { + key = message.getKeyBytes(); + } else if (message.hasOrderingKey()) { + key = message.getOrderingKey(); + } + // select a thread by message key to execute the runnable! + // that say, the message listener task with same order key + // will be executed by the same thread + ExecutorService executorService = executorProvider.getExecutor(key); + // executorService is a SingleThreadExecutor + executorService.execute(runnable); + }; +} + +private static MessageListenerExecutor getPartitionOrderdMessageListenerExecutor(String subscriptionName) { + ExecutorProvider executorProvider = new ExecutorProvider(10, subscriptionName + "listener-executor-"); + + return (message, runnable) -> { + // select a thread by partition topic name to execute the runnable! + // that say, the message listener task from the same partition topic + // will be executed by the same thread + ExecutorService executorService = executorProvider.getExecutor(message.getTopicName().getBytes()); + // executorService is a SingleThreadExecutor + executorService.execute(runnable); + }; +} + ``` ## Public-facing Changes From 23a9c779934e77579b2e90c19a80aca57ba52ed6 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Tue, 18 Jun 2024 23:03:29 +0800 Subject: [PATCH 06/10] refactor API --- pip/pip-359.md | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 3b25c6dcd3ce1..82de5e67f9355 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -112,7 +112,8 @@ to ensure that the messages of the same order key (or partition) are processed i .messageListener((c1, msg) -> { log.info("Received message [{}] in the listener", msg.getValue()); c1.acknowledgeAsync(msg); - }, keySharedExecutor) + }) + .messageListenerExecutor(keySharedExecutor) .subscribe(); @@ -126,7 +127,8 @@ to ensure that the messages of the same order key (or partition) are processed i .messageListener((c1, msg) -> { log.info("Received message [{}] in the listener", msg.getValue()); c1.acknowledgeAsync(msg); - }, partitionOrderedExecutor) + }) + .messageListenerExecutor(partitionOrderedExecutor) .subscribe(); // for example: out-of-order @@ -139,7 +141,8 @@ to ensure that the messages of the same order key (or partition) are processed i .messageListener((c1, msg) -> { log.info("Received message [{}] in the listener", msg.getValue()); c1.acknowledgeAsync(msg); - }, (message, runnable) -> executorService.execute(runnable)) + }) + .messageListenerExecutor((message, runnable) -> executorService.execute(runnable)) .subscribe(); } @@ -180,9 +183,9 @@ private static MessageListenerExecutor getPartitionOrderdMessageListenerExecutor ### Public API -1. Add the following optional overloaded function `messageListener` in `ConsumerBuilder`, to support specifying the messageListener executor +1. Add an optional config `messageListenerExecutor` in `ConsumerBuilder` ```java -ConsumerBuilder messageListener(@NonNull MessageListener messageListener, @NonNull MessageListenerExecutor messageListenerExecutor); +ConsumerBuilder messageListenerExecutor(MessageListenerExecutor messageListenerExecutor); ``` # Backward & Forward Compatibility From 58e38282b958648bb3cf6092cf4c46a6a12e2b50 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Tue, 18 Jun 2024 23:26:42 +0800 Subject: [PATCH 07/10] Why need an interface like --- pip/pip-359.md | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/pip/pip-359.md b/pip/pip-359.md index 82de5e67f9355..0be05dc4d1e78 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -93,6 +93,17 @@ public interface MessageListenerExecutor { } ``` +### Why need an interface like `MessageListenerExecutor` +Some people may wonder why not just use `java.util.concurrent.ExecutorService`, +but define an interface like `MessageListenerExecutor`. + +The reason is that: + +For sequential consumption scenarios, we need to ensure that messages with the same +key or the same partition are processed by the same thread to ensure order. If we +use `java.util.concurrent.ExecutorService`, we will not be able to make such guarantees, +because for ExecutorService, which thread to execute the task is not controlled by the user. + ### Interface implementation suggestions When implementing the `MessageListenerExecutor` interface, you should consider the following points. 1. if you need to ensure the order of message processing, From ae2c0f7cac67168eaf8ec38920ec9a372a9b3f6f Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Wed, 19 Jun 2024 01:31:56 +0800 Subject: [PATCH 08/10] add email voting thread link --- pip/pip-359.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 0be05dc4d1e78..5c6d997b5e563 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -205,4 +205,4 @@ You can do upgrading or reverting normally, no specified steps are needed to do. # Links * Mailing List discussion thread: https://lists.apache.org/thread/8nhqfdhkglsg5bgx6z7c1nho7z7l596l -* Mailing List voting thread: +* Mailing List voting thread: https://lists.apache.org/thread/oo3jdvq3b6bv6p4n7x7sdvypw4gp6hpk From 43d013ef6d9d84d7210042a2d83c126be99eb6c6 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Wed, 19 Jun 2024 01:37:56 +0800 Subject: [PATCH 09/10] update Design & Implementation Details --- pip/pip-359.md | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 5c6d997b5e563..1229a98055a48 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -72,14 +72,14 @@ we will open a new PIP in the future to do this ## Design & Implementation Details -Add an interface `MessageListenerExecutor`, responsible for executing message listener callback tasks. +1. Add an interface `MessageListenerExecutor`, responsible for executing message listener callback tasks. Users can customize the implementation to determine in which thread the message listener task is executed. For example, in the situation described in [Motivation](#motivation) part, users can implement the interface with an independent underlying thread pool to ensure that the message listener task of each consumer is executed in a separate thread. The caller would be responsible for the life cycle of the Executor, and it would be used only for this specific consumer. -```java -public interface MessageListenerExecutor { + ```java + public interface MessageListenerExecutor { /** * select a thread by message(if necessary, for example, @@ -90,8 +90,13 @@ public interface MessageListenerExecutor { * @param runnable the runnable to execute */ void execute(Message message, Runnable runnable); -} -``` + } + ``` +2. Add an optional config `messageListenerExecutor` in `ConsumerBuilder`, then +users can pass their implementations. + ```java + ConsumerBuilder messageListenerExecutor(MessageListenerExecutor messageListenerExecutor); + ``` ### Why need an interface like `MessageListenerExecutor` Some people may wonder why not just use `java.util.concurrent.ExecutorService`, From bfdcaa9b69c45810b2bee81b964cbeaf30ac8878 Mon Sep 17 00:00:00 2001 From: duanlinlin Date: Wed, 19 Jun 2024 16:28:12 +0800 Subject: [PATCH 10/10] update doc and add figure --- pip/pip-359.md | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/pip/pip-359.md b/pip/pip-359.md index 1229a98055a48..52a76193d6cf2 100644 --- a/pip/pip-359.md +++ b/pip/pip-359.md @@ -2,8 +2,9 @@ Implementation PR: [#22861](https://github.com/apache/pulsar/pull/22861) # Background knowledge -In the current Pulsar client versions, when using a Pulsar Consumer, we have two options to consume messages: -1. Synchronously, by calling `consumer.recieve()` +In the current Pulsar client versions, from the user's perspective, when using a Pulsar Consumer, +we have two main options to consume messages: +1. Pull mode, by calling `consumer.recieve()`(or `consumer.recieveAsync()`) ```java public class ConsumerExample { public static void main(String[] args) throws PulsarClientException { @@ -23,12 +24,13 @@ public class ConsumerExample { } ``` -2. Asynchronously, by registering a `MessageListener` interface, when building the Consumer. -When this method is used, we can't also use `consumer.receive()`. In the asynchronous way, -the MessageListener instance is called by the consumer, hence it is doing that with a thread -taken from its own internal `ExecutorService` (i.e. thread pool). The problem comes when we -build and use multiple Consumers from the same PulsarClient. It so happens that those consumers -will share the same thread pool to call the Message Listeners. One can be slower from the other. +2. Push mode, by registering a `MessageListener` interface, when building the Consumer. +When this method is used, we can't also use `consumer.receive()`(or `consumer.recieveAsync()`). +In the push mode, the MessageListener instance is called by the consumer, hence it is +doing that with a thread taken from its own internal `ExecutorService` (i.e. thread pool). +The problem comes when we build and use multiple Consumers from the same PulsarClient. It +so happens that those consumers will share the same thread pool to call the Message Listeners. +One can be slower from the other. ```java public class ConsumerExample { @@ -106,9 +108,10 @@ The reason is that: For sequential consumption scenarios, we need to ensure that messages with the same key or the same partition are processed by the same thread to ensure order. If we -use `java.util.concurrent.ExecutorService`, we will not be able to make such guarantees, +use `java.util.concurrent.ExecutorService`, refer to the following figure, we will not be able to make such guarantees, because for ExecutorService, which thread to execute the task is not controlled by the user. - +![](https://github.com/AuroraTwinkle/pulsar/assets/25919180/232854d6-01f2-4821-b2df-34d01dda1992) +![](https://github.com/AuroraTwinkle/pulsar/assets/25919180/204f5622-1e5a-4e73-b86b-15220bfb06d6) ### Interface implementation suggestions When implementing the `MessageListenerExecutor` interface, you should consider the following points. 1. if you need to ensure the order of message processing,