You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/06/29 02:32:05 UTC
[pulsar] branch branch-2.11 updated: [fix][client]Fix deadlock issue of consumer while using multiple IO threads (#20669)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new 4cf87a64e77 [fix][client]Fix deadlock issue of consumer while using multiple IO threads (#20669)
4cf87a64e77 is described below
commit 4cf87a64e77ba99a16b90ef7356d25905e5dd2b8
Author: Penghui Li <pe...@apache.org>
AuthorDate: Thu Jun 29 10:05:31 2023 +0800
[fix][client]Fix deadlock issue of consumer while using multiple IO threads (#20669)
---
.../pulsar/client/api/MultiTopicsConsumerTest.java | 23 ++++++++++++++
.../client/impl/MultiTopicsConsumerImpl.java | 37 +++++++++++-----------
2 files changed, 41 insertions(+), 19 deletions(-)
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
index 75b315413f5..25f0415ccda 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/client/api/MultiTopicsConsumerTest.java
@@ -24,6 +24,7 @@ import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.times;
import static org.mockito.Mockito.verify;
+import static org.testng.Assert.assertTrue;
import com.google.common.collect.Lists;
import java.util.HashMap;
import java.util.Map;
@@ -36,8 +37,10 @@ import java.util.concurrent.atomic.AtomicLong;
import lombok.Cleanup;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.impl.ClientBuilderImpl;
+import org.apache.pulsar.client.impl.MultiTopicsConsumerImpl;
import org.apache.pulsar.client.impl.PulsarClientImpl;
import org.apache.pulsar.client.impl.conf.ClientConfigurationData;
+import org.apache.pulsar.common.naming.TopicName;
import org.mockito.AdditionalAnswers;
import org.mockito.Mockito;
import org.slf4j.Logger;
@@ -65,6 +68,11 @@ public class MultiTopicsConsumerTest extends ProducerConsumerBase {
super.internalCleanup();
}
+ @Override
+ protected void customizeNewPulsarClientBuilder(ClientBuilder clientBuilder) {
+ clientBuilder.ioThreads(4).connectionsPerBroker(4);
+ }
+
// test that reproduces the issue https://github.com/apache/pulsar/issues/12024
// where closing the consumer leads to an endless receive loop
@Test
@@ -193,4 +201,19 @@ public class MultiTopicsConsumerTest extends ProducerConsumerBase {
}
Assert.assertEquals(numPartitions * numMessages, receivedCount);
}
+
+ @Test(invocationCount = 10, timeOut = 30000)
+ public void testMultipleIOThreads() throws PulsarAdminException, PulsarClientException {
+ final var topic = TopicName.get(newTopicName()).toString();
+ final var numPartitions = 100;
+ admin.topics().createPartitionedTopic(topic, numPartitions);
+ for (int i = 0; i < 100; i++) {
+ admin.topics().createNonPartitionedTopic(topic + "-" + i);
+ }
+ @Cleanup
+ final var consumer = pulsarClient.newConsumer(Schema.INT32).topicsPattern(topic + ".*")
+ .subscriptionName("sub").subscribe();
+ assertTrue(consumer instanceof MultiTopicsConsumerImpl);
+ assertTrue(consumer.isConnected());
+ }
}
diff --git a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
index 92551f2055b..15ae360ebf5 100644
--- a/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
+++ b/pulsar-client/src/main/java/org/apache/pulsar/client/impl/MultiTopicsConsumerImpl.java
@@ -89,7 +89,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
// sum of topicPartitions, simple topic has 1, partitioned topic equals to partition number.
AtomicInteger allTopicPartitionsNumber;
- private boolean paused = false;
+ private volatile boolean paused = false;
private final Object pauseMutex = new Object();
// timeout related to auto check and subscribe partition increasement
private volatile Timeout partitionsAutoUpdateTimeout = null;
@@ -1087,29 +1087,28 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
CompletableFuture<Consumer<T>> subFuture = new CompletableFuture<>();
- consumers.compute(topicName, (key, existingValue) -> {
- if (existingValue != null) {
- String errorMessage = String.format("[%s] Failed to subscribe for topic [%s] in topics consumer. "
- + "Topic is already being subscribed for in other thread.", topic, topicName);
- log.warn(errorMessage);
- subscribeResult.completeExceptionally(new PulsarClientException(errorMessage));
- return existingValue;
- } else {
- internalConfig.setStartPaused(paused);
- ConsumerImpl<T> newConsumer = createInternalConsumer(internalConfig, topicName,
- -1, subFuture, createIfDoesNotExist, schema);
-
- synchronized (pauseMutex) {
+ synchronized (pauseMutex) {
+ consumers.compute(topicName, (key, existingValue) -> {
+ if (existingValue != null) {
+ String errorMessage =
+ String.format("[%s] Failed to subscribe for topic [%s] in topics consumer. "
+ + "Topic is already being subscribed for in other thread.", topic, topicName);
+ log.warn(errorMessage);
+ subscribeResult.completeExceptionally(new PulsarClientException(errorMessage));
+ return existingValue;
+ } else {
+ internalConfig.setStartPaused(paused);
+ ConsumerImpl<T> newConsumer = createInternalConsumer(internalConfig, topicName,
+ -1, subFuture, createIfDoesNotExist, schema);
if (paused) {
newConsumer.pause();
} else {
newConsumer.resume();
}
+ return newConsumer;
}
- return newConsumer;
- }
- });
-
+ });
+ }
futureList = Collections.singletonList(subFuture);
}
@@ -1436,7 +1435,7 @@ public class MultiTopicsConsumerImpl<T> extends ConsumerBase<T> {
}
if (log.isDebugEnabled()) {
log.debug("[{}] create consumer {} for partitionName: {}",
- topicName, newConsumer.getTopic(), partitionName);
+ topicName, newConsumer.getTopic(), partitionName);
}
return subFuture;
})