From 416b2d9bd0c8607c55b822788f8e0e23e97dd0f9 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 19:25:06 +0800 Subject: [PATCH 1/8] Add test to reproduce stack overflow --- ...entDispatcherSingleActiveConsumerTest.java | 59 +++++++++++++++++++ 1 file changed, 59 insertions(+) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index dc6d451ed0fdf..4c67a5f094a8d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -18,7 +18,11 @@ */ package org.apache.pulsar.broker.service.persistent; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import lombok.Cleanup; import lombok.extern.slf4j.Slf4j; import org.apache.bookkeeper.mledger.AsyncCallbacks; @@ -26,12 +30,15 @@ import org.apache.bookkeeper.mledger.ManagedLedgerException; import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.pulsar.broker.BrokerTestUtil; +import org.apache.pulsar.broker.intercept.MockBrokerInterceptor; import org.apache.pulsar.broker.service.Consumer; +import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.api.proto.CommandSubscribe; +import org.apache.pulsar.common.naming.TopicName; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; @@ -42,11 +49,14 @@ @Slf4j @Test(groups = "broker-api") public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsumerBase { + private final Interceptor interceptor = new Interceptor(); + @BeforeClass(alwaysRun = true) @Override protected void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); + pulsar.getBrokerService().setInterceptor(interceptor); } @AfterClass(alwaysRun = true) @@ -129,4 +139,53 @@ public void testSkipReadEntriesFromCloseCursor() throws Exception { // Verify: the topic can be deleted successfully. admin.topics().delete(topicName, false); } + + @Test + public void testOverrideInactiveConsumer() throws Exception { + final var topic = "test-override-inactive-consumer"; + @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe(); + final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) + .toString()).get().orElseThrow()).getSubscription("sub").dispatcher; + Assert.assertEquals(dispatcher.getConsumers().size(), 1); + + // Generally `isActive` could only be false after `channelInactive` is called, setting it with false directly + // to avoid race condition. + final var latch = new CountDownLatch(1); + interceptor.latch.set(latch); + interceptor.injectCloseLatency.set(true); + // Simulate the real case because `channelInactive` is always called in the event loop thread + final var cnx = (ServerCnx) dispatcher.getConsumers().get(0).cnx(); + cnx.ctx().executor().execute(() -> { + try { + cnx.channelInactive(cnx.ctx()); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); + + final var mockConsumer = Mockito.mock(Consumer.class); + Assert.assertTrue(latch.await(1, TimeUnit.SECONDS)); + dispatcher.addConsumer(mockConsumer).get(); + } + + private static class Interceptor extends MockBrokerInterceptor { + + final AtomicBoolean injectCloseLatency = new AtomicBoolean(false); + final AtomicReference latch = new AtomicReference<>(); + + @Override + public void onConnectionClosed(ServerCnx cnx) { + if (injectCloseLatency.compareAndSet(true, false)) { + if (latch.get() != null) { + latch.get().countDown(); + } + try { + Thread.sleep(1000); + } catch (InterruptedException e) { + throw new RuntimeException(e); + } + } + super.onConnectionClosed(cnx); + } + } } From 6fd0974bcb37e0ef444f8a7717834b85123849e4 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 20:06:08 +0800 Subject: [PATCH 2/8] Fix race condition --- ...bstractDispatcherSingleActiveConsumer.java | 34 +++++++++++++++++-- ...entDispatcherSingleActiveConsumerTest.java | 2 +- 2 files changed, 33 insertions(+), 3 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java index baca6bf078cf0..cde08ac70aa4b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java @@ -28,6 +28,7 @@ import java.util.TreeMap; import java.util.concurrent.CompletableFuture; import java.util.concurrent.CopyOnWriteArrayList; +import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicIntegerFieldUpdater; import org.apache.bookkeeper.mledger.ManagedCursor; @@ -45,6 +46,7 @@ public abstract class AbstractDispatcherSingleActiveConsumer extends AbstractBaseDispatcher { + private static final int MAX_RETRY_COUNT_FOR_ADD_CONSUMER_RACE = 5; protected final String topicName; private volatile Consumer activeConsumer = null; protected final CopyOnWriteArrayList consumers; @@ -161,7 +163,21 @@ private NavigableMap makeHashRing(int consumerSize) { return Collections.unmodifiableNavigableMap(hashRing); } - public synchronized CompletableFuture addConsumer(Consumer consumer) { + public CompletableFuture addConsumer(Consumer consumer) { + return internalAddConsumer(consumer, 0); + } + + private synchronized CompletableFuture internalAddConsumer(Consumer consumer, int retryCount) { + if (retryCount >= MAX_RETRY_COUNT_FOR_ADD_CONSUMER_RACE) { + log.warn("[{}] The active consumer's connection is still inactive after all retries, remove {} by force", + getName(), consumer); + try { + removeConsumer(consumer); + } catch (BrokerServiceException e) { + // Ignore the exception because it could only be thrown when the consumer is already removed + log.warn("[{}] Failed to remove inactive consumer {}", getName(), e); + } + } if (IS_CLOSED_UPDATER.get(this) == TRUE) { log.warn("[{}] Dispatcher is already closed. Closing consumer {}", this.topicName, consumer); consumer.disconnect(); @@ -171,12 +187,26 @@ public synchronized CompletableFuture addConsumer(Consumer consumer) { if (subscriptionType == SubType.Exclusive && !consumers.isEmpty()) { Consumer actConsumer = getActiveConsumer(); if (actConsumer != null) { + final var callerThread = Thread.currentThread(); return actConsumer.cnx().checkConnectionLiveness().thenCompose(actConsumerStillAlive -> { if (actConsumerStillAlive.isEmpty() || actConsumerStillAlive.get()) { return FutureUtil.failedFuture(new ConsumerBusyException("Exclusive consumer is already" + " connected")); } else { - return addConsumer(consumer); + if (Thread.currentThread().equals(callerThread)) { + // A race condition happened in `ServerCnx#channelInactive` + // 1. `isActive` was set to false + // 2. `consumer.close()` is called + // We should wait for the + log.warn("[{}] race condition happened that cnx of the active consumer ({}) is inactive " + + "but it's not removed, retrying", getName(), actConsumer); + final var future = new CompletableFuture(); + CompletableFuture.delayedExecutor(100, TimeUnit.MILLISECONDS) + .execute(() -> future.complete(null)); + return future.thenCompose(__ -> addConsumer(consumer)); + } else { + return addConsumer(consumer); + } } }); } else { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index 4c67a5f094a8d..80e37814f6a68 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -180,7 +180,7 @@ public void onConnectionClosed(ServerCnx cnx) { latch.get().countDown(); } try { - Thread.sleep(1000); + Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); } From b5b2d4659789b8bf959d72ca124734ed0fa4bba8 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 20:27:41 +0800 Subject: [PATCH 3/8] Fix tests --- ...entDispatcherSingleActiveConsumerTest.java | 23 ++++++++++--------- 1 file changed, 12 insertions(+), 11 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index 80e37814f6a68..7b4dc1e352e40 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -18,6 +18,7 @@ */ package org.apache.pulsar.broker.service.persistent; +import java.util.Optional; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; @@ -42,24 +43,22 @@ import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; -import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j @Test(groups = "broker-api") public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsumerBase { - private final Interceptor interceptor = new Interceptor(); - @BeforeClass(alwaysRun = true) + @BeforeMethod(alwaysRun = true) @Override protected void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); - pulsar.getBrokerService().setInterceptor(interceptor); } - @AfterClass(alwaysRun = true) + @AfterMethod(alwaysRun = true) @Override protected void cleanup() throws Exception { super.internalCleanup(); @@ -142,6 +141,8 @@ public void testSkipReadEntriesFromCloseCursor() throws Exception { @Test public void testOverrideInactiveConsumer() throws Exception { + final var interceptor = new Interceptor(); + pulsar.getBrokerService().setInterceptor(interceptor); final var topic = "test-override-inactive-consumer"; @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe(); final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) @@ -163,9 +164,11 @@ public void testOverrideInactiveConsumer() throws Exception { } }); - final var mockConsumer = Mockito.mock(Consumer.class); + @Cleanup final var mockConsumer = Mockito.mock(Consumer.class); Assert.assertTrue(latch.await(1, TimeUnit.SECONDS)); dispatcher.addConsumer(mockConsumer).get(); + Assert.assertEquals(dispatcher.getConsumers().size(), 1); + Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0)); } private static class Interceptor extends MockBrokerInterceptor { @@ -176,16 +179,14 @@ private static class Interceptor extends MockBrokerInterceptor { @Override public void onConnectionClosed(ServerCnx cnx) { if (injectCloseLatency.compareAndSet(true, false)) { - if (latch.get() != null) { - latch.get().countDown(); - } + Optional.ofNullable(latch.get()).ifPresent(CountDownLatch::countDown); + latch.set(null); try { Thread.sleep(500); } catch (InterruptedException e) { throw new RuntimeException(e); } } - super.onConnectionClosed(cnx); } } } From ea909e65cf57ad4037d65cc8c60d69568e5d11d4 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 20:35:57 +0800 Subject: [PATCH 4/8] Improve the test to avoid calling channelInactive directly --- ...entDispatcherSingleActiveConsumerTest.java | 25 ++++++++----------- 1 file changed, 10 insertions(+), 15 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index 7b4dc1e352e40..f119aeb181a18 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -37,28 +37,29 @@ import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.api.proto.CommandSubscribe; import org.apache.pulsar.common.naming.TopicName; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; -import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeMethod; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @Slf4j @Test(groups = "broker-api") public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsumerBase { - @BeforeMethod(alwaysRun = true) + @BeforeClass(alwaysRun = true) @Override protected void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); } - @AfterMethod(alwaysRun = true) + @AfterClass(alwaysRun = true) @Override protected void cleanup() throws Exception { super.internalCleanup(); @@ -144,7 +145,8 @@ public void testOverrideInactiveConsumer() throws Exception { final var interceptor = new Interceptor(); pulsar.getBrokerService().setInterceptor(interceptor); final var topic = "test-override-inactive-consumer"; - @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe(); + @Cleanup final var client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe(); final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) .toString()).get().orElseThrow()).getSubscription("sub").dispatcher; Assert.assertEquals(dispatcher.getConsumers().size(), 1); @@ -154,21 +156,14 @@ public void testOverrideInactiveConsumer() throws Exception { final var latch = new CountDownLatch(1); interceptor.latch.set(latch); interceptor.injectCloseLatency.set(true); - // Simulate the real case because `channelInactive` is always called in the event loop thread - final var cnx = (ServerCnx) dispatcher.getConsumers().get(0).cnx(); - cnx.ctx().executor().execute(() -> { - try { - cnx.channelInactive(cnx.ctx()); - } catch (Exception e) { - throw new RuntimeException(e); - } - }); + final var future = client.closeAsync(); @Cleanup final var mockConsumer = Mockito.mock(Consumer.class); - Assert.assertTrue(latch.await(1, TimeUnit.SECONDS)); + Assert.assertTrue(latch.await(3, TimeUnit.SECONDS)); dispatcher.addConsumer(mockConsumer).get(); Assert.assertEquals(dispatcher.getConsumers().size(), 1); Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0)); + future.get(3, TimeUnit.SECONDS); } private static class Interceptor extends MockBrokerInterceptor { From 5fd8eff3d3c9759f8eb2380912c9d3ae1b2081e3 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 20:38:52 +0800 Subject: [PATCH 5/8] Revert "Improve the test to avoid calling channelInactive directly" This reverts commit ea909e65cf57ad4037d65cc8c60d69568e5d11d4. --- ...entDispatcherSingleActiveConsumerTest.java | 25 +++++++++++-------- 1 file changed, 15 insertions(+), 10 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index f119aeb181a18..7b4dc1e352e40 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -37,29 +37,28 @@ import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; -import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.api.proto.CommandSubscribe; import org.apache.pulsar.common.naming.TopicName; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; -import org.testng.annotations.AfterClass; -import org.testng.annotations.BeforeClass; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; import org.testng.annotations.Test; @Slf4j @Test(groups = "broker-api") public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsumerBase { - @BeforeClass(alwaysRun = true) + @BeforeMethod(alwaysRun = true) @Override protected void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); } - @AfterClass(alwaysRun = true) + @AfterMethod(alwaysRun = true) @Override protected void cleanup() throws Exception { super.internalCleanup(); @@ -145,8 +144,7 @@ public void testOverrideInactiveConsumer() throws Exception { final var interceptor = new Interceptor(); pulsar.getBrokerService().setInterceptor(interceptor); final var topic = "test-override-inactive-consumer"; - @Cleanup final var client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); - final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe(); + @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe(); final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) .toString()).get().orElseThrow()).getSubscription("sub").dispatcher; Assert.assertEquals(dispatcher.getConsumers().size(), 1); @@ -156,14 +154,21 @@ public void testOverrideInactiveConsumer() throws Exception { final var latch = new CountDownLatch(1); interceptor.latch.set(latch); interceptor.injectCloseLatency.set(true); - final var future = client.closeAsync(); + // Simulate the real case because `channelInactive` is always called in the event loop thread + final var cnx = (ServerCnx) dispatcher.getConsumers().get(0).cnx(); + cnx.ctx().executor().execute(() -> { + try { + cnx.channelInactive(cnx.ctx()); + } catch (Exception e) { + throw new RuntimeException(e); + } + }); @Cleanup final var mockConsumer = Mockito.mock(Consumer.class); - Assert.assertTrue(latch.await(3, TimeUnit.SECONDS)); + Assert.assertTrue(latch.await(1, TimeUnit.SECONDS)); dispatcher.addConsumer(mockConsumer).get(); Assert.assertEquals(dispatcher.getConsumers().size(), 1); Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0)); - future.get(3, TimeUnit.SECONDS); } private static class Interceptor extends MockBrokerInterceptor { From caf534f0d2d97fbe8755aef35d52966cbd367966 Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 20:41:51 +0800 Subject: [PATCH 6/8] Improve tests to avoid affecting other tests --- ...PersistentDispatcherSingleActiveConsumerTest.java | 12 +++++++----- 1 file changed, 7 insertions(+), 5 deletions(-) diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index 7b4dc1e352e40..4e531021beb13 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -37,28 +37,29 @@ import org.apache.pulsar.broker.service.Subscription; import org.apache.pulsar.client.api.Producer; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClient; import org.apache.pulsar.client.api.Schema; import org.apache.pulsar.common.api.proto.CommandSubscribe; import org.apache.pulsar.common.naming.TopicName; import org.awaitility.Awaitility; import org.mockito.Mockito; import org.testng.Assert; -import org.testng.annotations.AfterMethod; -import org.testng.annotations.BeforeMethod; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @Slf4j @Test(groups = "broker-api") public class PersistentDispatcherSingleActiveConsumerTest extends ProducerConsumerBase { - @BeforeMethod(alwaysRun = true) + @BeforeClass(alwaysRun = true) @Override protected void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); } - @AfterMethod(alwaysRun = true) + @AfterClass(alwaysRun = true) @Override protected void cleanup() throws Exception { super.internalCleanup(); @@ -144,7 +145,8 @@ public void testOverrideInactiveConsumer() throws Exception { final var interceptor = new Interceptor(); pulsar.getBrokerService().setInterceptor(interceptor); final var topic = "test-override-inactive-consumer"; - @Cleanup final var consumer = pulsarClient.newConsumer().topic(topic).subscriptionName("sub").subscribe(); + @Cleanup final var client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); + @Cleanup final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe(); final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) .toString()).get().orElseThrow()).getSubscription("sub").dispatcher; Assert.assertEquals(dispatcher.getConsumers().size(), 1); From 4c7d5512d7bd26257d289ecd768a5b78a05caead Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 21:01:40 +0800 Subject: [PATCH 7/8] Fix retry logic and fail when the count is reached --- ...bstractDispatcherSingleActiveConsumer.java | 19 ++++------- ...entDispatcherSingleActiveConsumerTest.java | 33 +++++++++++++++---- 2 files changed, 33 insertions(+), 19 deletions(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java index cde08ac70aa4b..cdd5c235f579b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java @@ -168,16 +168,6 @@ public CompletableFuture addConsumer(Consumer consumer) { } private synchronized CompletableFuture internalAddConsumer(Consumer consumer, int retryCount) { - if (retryCount >= MAX_RETRY_COUNT_FOR_ADD_CONSUMER_RACE) { - log.warn("[{}] The active consumer's connection is still inactive after all retries, remove {} by force", - getName(), consumer); - try { - removeConsumer(consumer); - } catch (BrokerServiceException e) { - // Ignore the exception because it could only be thrown when the consumer is already removed - log.warn("[{}] Failed to remove inactive consumer {}", getName(), e); - } - } if (IS_CLOSED_UPDATER.get(this) == TRUE) { log.warn("[{}] Dispatcher is already closed. Closing consumer {}", this.topicName, consumer); consumer.disconnect(); @@ -192,6 +182,11 @@ private synchronized CompletableFuture internalAddConsumer(Consumer consum if (actConsumerStillAlive.isEmpty() || actConsumerStillAlive.get()) { return FutureUtil.failedFuture(new ConsumerBusyException("Exclusive consumer is already" + " connected")); + } else if (retryCount >= MAX_RETRY_COUNT_FOR_ADD_CONSUMER_RACE) { + log.warn("[{}] The active consumer's connection is still inactive after all retries {}, skip " + + "adding new consumer {}", getName(), actConsumer, consumer); + return FutureUtil.failedFuture(new ConsumerBusyException("Exclusive consumer is already" + + " connected after " + MAX_RETRY_COUNT_FOR_ADD_CONSUMER_RACE + " attempts")); } else { if (Thread.currentThread().equals(callerThread)) { // A race condition happened in `ServerCnx#channelInactive` @@ -203,9 +198,9 @@ private synchronized CompletableFuture internalAddConsumer(Consumer consum final var future = new CompletableFuture(); CompletableFuture.delayedExecutor(100, TimeUnit.MILLISECONDS) .execute(() -> future.complete(null)); - return future.thenCompose(__ -> addConsumer(consumer)); + return future.thenCompose(__ -> internalAddConsumer(consumer, retryCount + 1)); } else { - return addConsumer(consumer); + return internalAddConsumer(consumer, retryCount + 1); } } }); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java index 4e531021beb13..466f55436ea40 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherSingleActiveConsumerTest.java @@ -20,6 +20,7 @@ import java.util.Optional; import java.util.concurrent.CountDownLatch; +import java.util.concurrent.ExecutionException; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; @@ -32,6 +33,7 @@ import org.apache.bookkeeper.mledger.impl.ManagedCursorImpl; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.intercept.MockBrokerInterceptor; +import org.apache.pulsar.broker.service.BrokerServiceException; import org.apache.pulsar.broker.service.Consumer; import org.apache.pulsar.broker.service.ServerCnx; import org.apache.pulsar.broker.service.Subscription; @@ -46,6 +48,7 @@ import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; +import org.testng.annotations.DataProvider; import org.testng.annotations.Test; @Slf4j @@ -140,11 +143,16 @@ public void testSkipReadEntriesFromCloseCursor() throws Exception { admin.topics().delete(topicName, false); } - @Test - public void testOverrideInactiveConsumer() throws Exception { + @DataProvider + public static Object[][] closeDelayMs() { + return new Object[][] { { 500 }, { 2000 } }; + } + + @Test(dataProvider = "closeDelayMs") + public void testOverrideInactiveConsumer(long closeDelayMs) throws Exception { final var interceptor = new Interceptor(); pulsar.getBrokerService().setInterceptor(interceptor); - final var topic = "test-override-inactive-consumer"; + final var topic = "test-override-inactive-consumer-" + closeDelayMs; @Cleanup final var client = PulsarClient.builder().serviceUrl(pulsar.getBrokerServiceUrl()).build(); @Cleanup final var consumer = client.newConsumer().topic(topic).subscriptionName("sub").subscribe(); final var dispatcher = ((PersistentTopic) pulsar.getBrokerService().getTopicIfExists(TopicName.get(topic) @@ -156,6 +164,7 @@ public void testOverrideInactiveConsumer() throws Exception { final var latch = new CountDownLatch(1); interceptor.latch.set(latch); interceptor.injectCloseLatency.set(true); + interceptor.delayMs = closeDelayMs; // Simulate the real case because `channelInactive` is always called in the event loop thread final var cnx = (ServerCnx) dispatcher.getConsumers().get(0).cnx(); cnx.ctx().executor().execute(() -> { @@ -168,15 +177,25 @@ public void testOverrideInactiveConsumer() throws Exception { @Cleanup final var mockConsumer = Mockito.mock(Consumer.class); Assert.assertTrue(latch.await(1, TimeUnit.SECONDS)); - dispatcher.addConsumer(mockConsumer).get(); - Assert.assertEquals(dispatcher.getConsumers().size(), 1); - Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0)); + if (closeDelayMs < 1000) { + dispatcher.addConsumer(mockConsumer).get(); + Assert.assertEquals(dispatcher.getConsumers().size(), 1); + Assert.assertSame(mockConsumer, dispatcher.getConsumers().get(0)); + } else { + try { + dispatcher.addConsumer(mockConsumer).get(); + Assert.fail(); + } catch (ExecutionException e) { + Assert.assertTrue(e.getCause() instanceof BrokerServiceException.ConsumerBusyException); + } + } } private static class Interceptor extends MockBrokerInterceptor { final AtomicBoolean injectCloseLatency = new AtomicBoolean(false); final AtomicReference latch = new AtomicReference<>(); + long delayMs = 500; @Override public void onConnectionClosed(ServerCnx cnx) { @@ -184,7 +203,7 @@ public void onConnectionClosed(ServerCnx cnx) { Optional.ofNullable(latch.get()).ifPresent(CountDownLatch::countDown); latch.set(null); try { - Thread.sleep(500); + Thread.sleep(delayMs); } catch (InterruptedException e) { throw new RuntimeException(e); } From 37e4815ea8b9de9cd27b237707f4cbf76ff8e25d Mon Sep 17 00:00:00 2001 From: Yunze Xu Date: Mon, 3 Nov 2025 21:03:16 +0800 Subject: [PATCH 8/8] Add comments --- .../broker/service/AbstractDispatcherSingleActiveConsumer.java | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java index cdd5c235f579b..792b75f2896d9 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractDispatcherSingleActiveConsumer.java @@ -192,7 +192,7 @@ private synchronized CompletableFuture internalAddConsumer(Consumer consum // A race condition happened in `ServerCnx#channelInactive` // 1. `isActive` was set to false // 2. `consumer.close()` is called - // We should wait for the + // We should wait until the consumer is closed, retry for some times log.warn("[{}] race condition happened that cnx of the active consumer ({}) is inactive " + "but it's not removed, retrying", getName(), actConsumer); final var future = new CompletableFuture();