-
Notifications
You must be signed in to change notification settings - Fork 3.6k
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
[improve][client]PIP-359:Support custom message listener executor for…
… specific subscription (#22861) Co-authored-by: duanlinlin <[email protected]> [PIP-359](#22902) Support custom message listener thread pool for specific subscription, avoid individual subscription listener consuming too much time leading to higher consumption delay in other subscriptions. <!-- ### Contribution Checklist - PR title format should be *[type][component] summary*. For details, see *[Guideline - Pulsar PR Naming Convention](https://pulsar.apache.org/contribute/develop-semantic-title/)*. - Fill out the template below to describe the changes contributed by the pull request. That will give reviewers the context they need to do the review. - Each pull request should address only one issue, not mix up code from multiple issues. - Each commit in the pull request has a meaningful commit message - Once all items of the checklist are addressed, remove the above text and this checklist, leaving only the filled out template below. --> <!-- Either this PR fixes an issue, --> <!-- or this PR is one task of an issue --> <!-- If the PR belongs to a PIP, please add the PIP link here --> <!-- Details of when a PIP is required and how the PIP process work, please see: https://github.com/apache/pulsar/blob/master/pip/README.md --> ### Motivation In our scenario, there is a centralized message proxy service, this service will use the same PulsarClient instance to create a lot of subscription groups to consume many topics and cache messages locally.Then the business will pull messages from the cache of the proxy service. It seems that there is no problem, but during use, we found that when the message processing time of several consumer groups (listener mode) is very high, it almost affects all consumer groups responsible for the proxy service, causing a large number of message delays. By analyzing the source code, we found that by default, all consumer instances created from the same PulsarClient will share a thread pool 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 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 thread pool with consumer latitudes to avoid mutual influence between different consumers. <!-- Explain here the context, and why you're making that change. What is the problem you're trying to solve. --> ### Modifications Support custom message listener thread pool for specific subscription. <!-- Describe the modifications you've done. -->
- Loading branch information
1 parent
6dd7c59
commit 10f4e02
Showing
6 changed files
with
280 additions
and
10 deletions.
There are no files selected for viewing
193 changes: 193 additions & 0 deletions
193
pulsar-broker/src/test/java/org/apache/pulsar/client/api/MessageListenerExecutorTest.java
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,193 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.client.api; | ||
|
|
||
| import static org.testng.Assert.assertTrue; | ||
| import com.google.common.util.concurrent.Uninterruptibles; | ||
| import java.time.Duration; | ||
| import java.util.ArrayList; | ||
| import java.util.List; | ||
| import java.util.concurrent.CompletableFuture; | ||
| import java.util.concurrent.CountDownLatch; | ||
| import java.util.concurrent.ExecutorService; | ||
| import java.util.concurrent.Executors; | ||
| import java.util.concurrent.Future; | ||
| import java.util.concurrent.atomic.AtomicLong; | ||
| import lombok.Cleanup; | ||
| import org.apache.pulsar.client.util.ExecutorProvider; | ||
| import org.apache.pulsar.common.naming.TopicName; | ||
| import org.slf4j.Logger; | ||
| import org.slf4j.LoggerFactory; | ||
| import org.testng.Assert; | ||
| import org.testng.annotations.AfterClass; | ||
| import org.testng.annotations.BeforeClass; | ||
| import org.testng.annotations.Test; | ||
|
|
||
| @Test(groups = "broker-api") | ||
| public class MessageListenerExecutorTest extends ProducerConsumerBase { | ||
| private static final Logger log = LoggerFactory.getLogger(MessageListenerExecutorTest.class); | ||
|
|
||
| @BeforeClass(alwaysRun = true) | ||
| @Override | ||
| protected void setup() throws Exception { | ||
| super.internalSetup(); | ||
| super.producerBaseSetup(); | ||
| } | ||
|
|
||
| @AfterClass(alwaysRun = true) | ||
| @Override | ||
| protected void cleanup() throws Exception { | ||
| super.internalCleanup(); | ||
| } | ||
|
|
||
| @Override | ||
| protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) { | ||
| // Set listenerThreads to 1 to reproduce the pr more easily in #22861 | ||
| clientBuilder.listenerThreads(1); | ||
| } | ||
|
|
||
| @Test | ||
| public void testConsumerMessageListenerExecutorIsolation() throws Exception { | ||
| log.info("-- Starting {} test --", methodName); | ||
|
|
||
| @Cleanup("shutdownNow") | ||
| ExecutorService executor = Executors.newCachedThreadPool(); | ||
| List<CompletableFuture<Long>> maxConsumeDelayWithDisableIsolationFutures = new ArrayList<>(); | ||
| int loops = 5; | ||
| long consumeSleepTimeMs = 10000; | ||
| for (int i = 0; i < loops; i++) { | ||
| // The first consumer will consume messages with sleep block 1s, | ||
| // and the others will consume messages without sleep block. | ||
| // The maxConsumeDelayWithDisableIsolation of all consumers | ||
| // should be greater than sleepTimeMs cause by disable MessageListenerExecutor. | ||
| CompletableFuture<Long> maxConsumeDelayFuture = startConsumeAndComputeMaxConsumeDelay( | ||
| "persistent://my-property/my-ns/testConsumerMessageListenerDisableIsolation-" + i, | ||
| "my-sub-testConsumerMessageListenerDisableIsolation-" + i, | ||
| i == 0 ? Duration.ofMillis(consumeSleepTimeMs) : Duration.ofMillis(0), | ||
| false, | ||
| executor); | ||
| maxConsumeDelayWithDisableIsolationFutures.add(maxConsumeDelayFuture); | ||
| } | ||
|
|
||
| // ensure all consumers consume messages delay more than consumeSleepTimeMs | ||
| boolean allDelayMoreThanConsumeSleepTimeMs = maxConsumeDelayWithDisableIsolationFutures.stream() | ||
| .map(CompletableFuture::join) | ||
| .allMatch(delay -> delay > consumeSleepTimeMs); | ||
| assertTrue(allDelayMoreThanConsumeSleepTimeMs); | ||
|
|
||
| List<CompletableFuture<Long>> maxConsumeDelayWhitEnableIsolationFutures = new ArrayList<>(); | ||
| for (int i = 0; i < loops; i++) { | ||
| // The first consumer will consume messages with sleep block 1s, | ||
| // and the others will consume messages without sleep block. | ||
| // The maxConsumeDelayWhitEnableIsolation of the first consumer | ||
| // should be greater than sleepTimeMs, and the others should be | ||
| // less than sleepTimeMs, cause by enable MessageListenerExecutor. | ||
| CompletableFuture<Long> maxConsumeDelayFuture = startConsumeAndComputeMaxConsumeDelay( | ||
| "persistent://my-property/my-ns/testConsumerMessageListenerEnableIsolation-" + i, | ||
| "my-sub-testConsumerMessageListenerEnableIsolation-" + i, | ||
| i == 0 ? Duration.ofMillis(consumeSleepTimeMs) : Duration.ofMillis(0), | ||
| true, | ||
| executor); | ||
| maxConsumeDelayWhitEnableIsolationFutures.add(maxConsumeDelayFuture); | ||
| } | ||
|
|
||
| assertTrue(maxConsumeDelayWhitEnableIsolationFutures.get(0).join() > consumeSleepTimeMs); | ||
| boolean remainingAlmostNoDelay = maxConsumeDelayWhitEnableIsolationFutures.stream() | ||
| .skip(1) | ||
| .map(CompletableFuture::join) | ||
| .allMatch(delay -> delay < 1000); | ||
| assertTrue(remainingAlmostNoDelay); | ||
|
|
||
| log.info("-- Exiting {} test --", methodName); | ||
| } | ||
|
|
||
| private CompletableFuture<Long> startConsumeAndComputeMaxConsumeDelay(String topic, String subscriptionName, | ||
| Duration consumeSleepTime, | ||
| boolean enableMessageListenerExecutorIsolation, | ||
| ExecutorService executorService) | ||
| throws Exception { | ||
| int numMessages = 2; | ||
| final CountDownLatch latch = new CountDownLatch(numMessages); | ||
| int numPartitions = 50; | ||
| TopicName nonIsolationTopicName = TopicName.get(topic); | ||
| admin.topics().createPartitionedTopic(nonIsolationTopicName.toString(), numPartitions); | ||
|
|
||
| AtomicLong maxConsumeDelay = new AtomicLong(-1); | ||
| ConsumerBuilder<Long> consumerBuilder = | ||
| pulsarClient.newConsumer(Schema.INT64) | ||
| .topic(nonIsolationTopicName.toString()) | ||
| .subscriptionName(subscriptionName) | ||
| .messageListener((c1, msg) -> { | ||
| Assert.assertNotNull(msg, "Message cannot be null"); | ||
| log.debug("Received message [{}] in the listener", msg.getValue()); | ||
| c1.acknowledgeAsync(msg); | ||
| maxConsumeDelay.set(Math.max(maxConsumeDelay.get(), | ||
| System.currentTimeMillis() - msg.getValue())); | ||
| if (consumeSleepTime.toMillis() > 0) { | ||
| Uninterruptibles.sleepUninterruptibly(consumeSleepTime); | ||
| } | ||
| latch.countDown(); | ||
| }); | ||
|
|
||
| ExecutorService executor = Executors.newSingleThreadExecutor( | ||
| new ExecutorProvider.ExtendedThreadFactory(subscriptionName + "listener-executor-", true)); | ||
| if (enableMessageListenerExecutorIsolation) { | ||
| consumerBuilder.messageListenerExecutor((message, runnable) -> executor.execute(runnable)); | ||
| } | ||
|
|
||
| Consumer<Long> consumer = consumerBuilder.subscribe(); | ||
| ProducerBuilder<Long> producerBuilder = pulsarClient.newProducer(Schema.INT64) | ||
| .topic(nonIsolationTopicName.toString()); | ||
|
|
||
| Producer<Long> producer = producerBuilder.create(); | ||
| List<Future<MessageId>> futures = new ArrayList<>(); | ||
|
|
||
| // Asynchronously produce messages | ||
| for (int i = 0; i < numMessages; i++) { | ||
| Future<MessageId> future = producer.sendAsync(System.currentTimeMillis()); | ||
| futures.add(future); | ||
| } | ||
|
|
||
| log.info("Waiting for async publish to complete"); | ||
| for (Future<MessageId> future : futures) { | ||
| future.get(); | ||
| } | ||
|
|
||
| CompletableFuture<Long> maxDelayFuture = new CompletableFuture<>(); | ||
|
|
||
| CompletableFuture.runAsync(() -> { | ||
| try { | ||
| latch.await(); | ||
| } catch (InterruptedException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| }, executorService).whenCompleteAsync((v, ex) -> { | ||
| maxDelayFuture.complete(maxConsumeDelay.get()); | ||
| try { | ||
| producer.close(); | ||
| consumer.close(); | ||
| executor.shutdownNow(); | ||
| } catch (PulsarClientException e) { | ||
| throw new RuntimeException(e); | ||
| } | ||
| }); | ||
|
|
||
| return maxDelayFuture; | ||
| } | ||
| } |
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
43 changes: 43 additions & 0 deletions
43
pulsar-client-api/src/main/java/org/apache/pulsar/client/api/MessageListenerExecutor.java
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
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,43 @@ | ||
| /* | ||
| * Licensed to the Apache Software Foundation (ASF) under one | ||
| * or more contributor license agreements. See the NOTICE file | ||
| * distributed with this work for additional information | ||
| * regarding copyright ownership. The ASF licenses this file | ||
| * to you under the Apache License, Version 2.0 (the | ||
| * "License"); you may not use this file except in compliance | ||
| * with the License. You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, | ||
| * software distributed under the License is distributed on an | ||
| * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY | ||
| * KIND, either express or implied. See the License for the | ||
| * specific language governing permissions and limitations | ||
| * under the License. | ||
| */ | ||
| package org.apache.pulsar.client.api; | ||
|
|
||
| /** | ||
| * Interface for providing service to execute message listeners. | ||
| */ | ||
| public interface MessageListenerExecutor { | ||
|
|
||
| /** | ||
| * select a thread by message to execute the runnable! | ||
| * <p> | ||
| * Suggestions: | ||
| * <p> | ||
| * 1. The message listener task will be submitted to this executor for execution, | ||
| * so the implementations of this interface should carefully consider execution | ||
| * order if sequential consumption is required. | ||
| * </p> | ||
| * <p> | ||
| * 2. The users should release resources(e.g. threads) of the executor after closing | ||
| * the consumer to avoid leaks. | ||
| * </p> | ||
| * @param message the message | ||
| * @param runnable the runnable to execute, that is, the message listener task | ||
| */ | ||
| void execute(Message<?> message, Runnable runnable); | ||
| } |
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
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
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