You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2026/10/19 21:18:26 UTC
(pulsar) 07/16: [fix][broker] Fence non-persistent subscriptions before asynchronous … (#26991)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch branch-5.0
in repository https://gitbox.apache.org/repos/asf/pulsar.git
commit 450a9197f0b75b6779be75439c80a9c6d3729e97
Author: Jiwei Guo <te...@apache.org>
AuthorDate: Mon Oct 19 20:51:35 2026 +0800
[fix][broker] Fence non-persistent subscriptions before asynchronous … (#26991)
(cherry picked from commit 6cbe4ac4f6469dc8ffc6ee0079355a1119effff3)
---
.../nonpersistent/NonPersistentSubscription.java | 3 ++
.../nonpersistent/NonPersistentTopicTest.java | 63 ++++++++++++++++++++++
2 files changed, 66 insertions(+)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
index 23ffc1e9e72d..a0f15d9e09c4 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -189,6 +189,9 @@ public class NonPersistentSubscription extends AbstractSubscription {
msgOutFromRemovedConsumer.add(stats.msgOutCounter);
// Unsubscribe when all the consumers disconnected.
if (dispatcher != null && CollectionUtils.isEmpty(dispatcher.getConsumers())) {
+ // Fence under the same lock as addConsumer before queuing removal, so a reconnect cannot
+ // attach to this subscription while it is being removed from the topic.
+ IS_FENCED_UPDATER.set(this, TRUE);
topic.unsubscribe(subName);
}
diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopicTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopicTest.java
index d09f3068c2f0..c7d665de05ea 100644
--- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopicTest.java
+++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopicTest.java
@@ -20,6 +20,7 @@ package org.apache.pulsar.broker.service.nonpersistent;
import static org.testng.Assert.assertEquals;
import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertThrows;
import static org.testng.Assert.assertTrue;
import java.lang.reflect.Field;
@@ -30,6 +31,7 @@ import java.util.concurrent.CompletableFuture;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
import org.apache.pulsar.broker.service.AbstractTopic;
+import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionFencedException;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.service.PulsarCommandSender;
import org.apache.pulsar.broker.service.SubscriptionOption;
@@ -338,6 +340,67 @@ public class NonPersistentTopicTest extends BrokerTestBase {
}
+ @Test
+ public void testReconnectWhileSubscriptionRemovalIsPending() throws Exception {
+ String topicName = "non-persistent://prop/ns-abc/reconnect-" + UUID.randomUUID();
+ String subName = "sub";
+ @Cleanup
+ Producer<String> producer = pulsarClient.newProducer(Schema.STRING)
+ .topic(topicName).enableBatching(false).create();
+ NonPersistentTopic topic = (NonPersistentTopic) pulsar.getBrokerService()
+ .getTopicReference(topicName).orElseThrow();
+ NonPersistentTopic spyTopic = Mockito.spy(topic);
+ pulsar.getBrokerService().getTopics().put(topicName,
+ CompletableFuture.completedFuture(Optional.of(spyTopic)));
+ NonPersistentSubscription subscription = Mockito.spy(
+ new NonPersistentSubscription(spyTopic, subName, Collections.emptyMap()));
+ spyTopic.getSubscriptions().put(subName, subscription);
+
+ CompletableFuture<Void> removalQueued = new CompletableFuture<>();
+ CompletableFuture<Void> allowRemoval = new CompletableFuture<>();
+ Mockito.doAnswer(inv -> {
+ CompletableFuture<Void> removal = allowRemoval.thenCompose(__ -> topic.unsubscribe(subName));
+ removalQueued.complete(null);
+ return removal;
+ }).when(spyTopic).unsubscribe(subName);
+
+ try {
+ @Cleanup
+ Consumer<String> first = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionName(subName).subscriptionMode(SubscriptionMode.NonDurable).subscribe();
+ first.close();
+ removalQueued.get(10, TimeUnit.SECONDS);
+
+ // Make the reconnect reach the old subscription before the queued removal can execute.
+ CompletableFuture<Throwable> addAttempt = new CompletableFuture<>();
+ Mockito.doAnswer(inv -> {
+ @SuppressWarnings("unchecked")
+ CompletableFuture<Void> result = (CompletableFuture<Void>) inv.callRealMethod();
+ result.whenComplete((__, error) -> addAttempt.complete(error));
+ return result;
+ }).when(subscription).addConsumer(Mockito.any());
+ CompletableFuture<Consumer<String>> reconnect = pulsarClient.newConsumer(Schema.STRING).topic(topicName)
+ .subscriptionName(subName).subscriptionMode(SubscriptionMode.NonDurable).subscribeAsync();
+
+ assertTrue(addAttempt.get(10, TimeUnit.SECONDS) instanceof SubscriptionFencedException);
+ assertTrue(subscription.getConsumers().isEmpty());
+
+ allowRemoval.complete(null);
+ @Cleanup
+ Consumer<String> second = reconnect.get(30, TimeUnit.SECONDS);
+ assertTrue(second.isConnected());
+ assertNotSame(spyTopic.getSubscription(subName), subscription);
+ assertEquals(admin.topics().getStats(topicName).getSubscriptions().get(subName).getConsumers().size(), 1);
+
+ producer.send("after-reconnect");
+ Message<String> message = second.receive(10, TimeUnit.SECONDS);
+ assertNotNull(message);
+ assertEquals(message.getValue(), "after-reconnect");
+ } finally {
+ allowRemoval.complete(null);
+ }
+ }
+
@Test
public void testRemoveProducerOnNonPersistentTopic() throws Exception {
final String topicName = "non-persistent://prop/ns-abc/topic_" + UUID.randomUUID();