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/22 08:32:06 UTC
(pulsar) 11/12: [fix][broker] Fix non-persistent subscribe cleanup during migration (#27003)
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 d029efdfa42b7de83017a211b9c30b7c79a939c0
Author: sinan liu <li...@gmail.com>
AuthorDate: Thu Oct 22 09:53:17 2026 +0800
[fix][broker] Fix non-persistent subscribe cleanup during migration (#27003)
(cherry picked from commit 327e41bf4c7d899cb8b66232b2061ad2ce442615)
---
.../service/nonpersistent/NonPersistentTopic.java | 27 ++---
.../nonpersistent/NonPersistentTopicTest.java | 131 +++++++++++++++++++--
2 files changed, 132 insertions(+), 26 deletions(-)
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
index fc8bd5b5e496..9a03e50d15c7 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java
@@ -56,6 +56,7 @@ import org.apache.pulsar.broker.service.BrokerServiceException.NotAllowedExcepti
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicBusyException;
import org.apache.pulsar.broker.service.BrokerServiceException.TopicFencedException;
+import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.BrokerServiceException.UnsupportedVersionException;
import org.apache.pulsar.broker.service.Consumer;
import org.apache.pulsar.broker.service.Dispatcher;
@@ -340,24 +341,18 @@ public class NonPersistentTopic extends AbstractTopic implements Topic, TopicPol
NonPersistentSubscription subscription = subscriptions.computeIfAbsent(subscriptionName,
name -> new NonPersistentSubscription(this, subscriptionName, subscriptionProperties));
- Consumer consumer = new Consumer(subscription, subType, topic, consumerId, priorityLevel, consumerName,
- false, cnx, cnx.getAuthRole(), metadata, readCompacted, keySharedMeta, MessageId.latest,
- DEFAULT_CONSUMER_EPOCH, schemaType);
- consumer.checkAndApplyTopicMigrationAsync().thenCompose(migrated -> {
+ CompletableFuture<Boolean> migrationCheck = subscription.isSubscriptionMigrated()
+ ? isClusterMigrationEnabled(brokerService.getPulsar(), topic)
+ : CompletableFuture.completedFuture(false);
+ migrationCheck.thenCompose(migrated -> {
if (migrated) {
- // The topic is migrated: checkAndApplyTopicMigrationAsync() has already sent the
- // TopicMigrated redirect and disconnected the consumer (which also released the usage
- // count taken by handleConsumerAdded above). Skip addConsumerToSubscription so the consumer
- // is never attached to the subscription on the old cluster. Sequencing the migration check
- // through thenCompose (instead of the previous fire-and-forget thenAccept that raced with
- // addConsumerToSubscription) is what guarantees the consumer is not added once migrated.
- log.info()
- .attr("subscription", subscriptionName)
- .attr("consumerId", consumerId)
- .log("Skipped subscription on migrated topic; consumer was redirected");
- future.complete(consumer);
- return CompletableFuture.<Void>completedFuture(null);
+ // Roll back the subscribe usage count through the failure path. ServerCnx redirects the
+ // request without closing a consumer that was never attached to the dispatcher.
+ return CompletableFuture.<Void>failedFuture(new TopicMigratedException("Topic is migrated"));
}
+ Consumer consumer = new Consumer(subscription, subType, topic, consumerId, priorityLevel, consumerName,
+ false, cnx, cnx.getAuthRole(), metadata, readCompacted, keySharedMeta, MessageId.latest,
+ DEFAULT_CONSUMER_EPOCH, schemaType);
return addConsumerToSubscription(subscription, consumer).thenRun(() -> {
if (!cnx.isActive()) {
try {
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 c7d665de05ea..b3b6b2eed813 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
@@ -18,7 +18,9 @@
*/
package org.apache.pulsar.broker.service.nonpersistent;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertFalse;
import static org.testng.Assert.assertNotNull;
import static org.testng.Assert.assertNotSame;
import static org.testng.Assert.assertThrows;
@@ -28,23 +30,34 @@ import java.util.Collections;
import java.util.Optional;
import java.util.UUID;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import lombok.Cleanup;
+import org.apache.pulsar.broker.resources.PulsarResources;
import org.apache.pulsar.broker.service.AbstractTopic;
+import org.apache.pulsar.broker.service.BrokerService;
import org.apache.pulsar.broker.service.BrokerServiceException.SubscriptionFencedException;
+import org.apache.pulsar.broker.service.BrokerServiceException.TopicMigratedException;
import org.apache.pulsar.broker.service.BrokerTestBase;
import org.apache.pulsar.broker.service.PulsarCommandSender;
import org.apache.pulsar.broker.service.SubscriptionOption;
import org.apache.pulsar.broker.service.TransportCnx;
import org.apache.pulsar.client.admin.PulsarAdminException;
import org.apache.pulsar.client.api.Consumer;
+import org.apache.pulsar.client.api.InjectedClientCnxClientBuilder;
import org.apache.pulsar.client.api.Message;
import org.apache.pulsar.client.api.Producer;
+import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.Schema;
import org.apache.pulsar.client.api.SubscriptionMode;
import org.apache.pulsar.client.api.SubscriptionType;
+import org.apache.pulsar.client.impl.ClientBuilderImpl;
+import org.apache.pulsar.client.impl.ClientCnx;
+import org.apache.pulsar.client.impl.metrics.InstrumentProvider;
import org.apache.pulsar.common.api.proto.CommandSubscribe;
+import org.apache.pulsar.common.api.proto.CommandTopicMigrated;
+import org.apache.pulsar.common.api.proto.CommandTopicMigrated.ResourceType;
import org.apache.pulsar.common.naming.TopicName;
import org.apache.pulsar.common.policies.data.ClusterPolicies.ClusterUrl;
import org.apache.pulsar.common.policies.data.TopicStats;
@@ -52,11 +65,17 @@ import org.awaitility.Awaitility;
import org.mockito.Mockito;
import org.testng.annotations.AfterMethod;
import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.DataProvider;
import org.testng.annotations.Test;
@Test(groups = "broker")
public class NonPersistentTopicTest extends BrokerTestBase {
+ @Override
+ protected BrokerService customizeNewBrokerService(BrokerService brokerService) {
+ return Mockito.spy(brokerService);
+ }
+
@BeforeMethod(alwaysRun = true)
@Override
protected void setup() throws Exception {
@@ -140,8 +159,8 @@ public class NonPersistentTopicTest extends BrokerTestBase {
* {@code addConsumerToSubscription}, and the consumer must NOT be attached to the subscription on the
* old cluster. The previous code ran {@code getMigratedClusterUrlAsync().thenAccept(consumer::topicMigrated)}
* concurrently with {@code addConsumerToSubscription}, so the consumer could be added before the redirect
- * and disconnect ran. The fix sequences the check through {@link org.apache.pulsar.broker.service.Consumer
- * #checkAndApplyTopicMigrationAsync()} and skips the add when migrated.
+ * and disconnect ran. The migration check must fail the subscribe before attaching a consumer, so that
+ * the subscribe failure path releases the usage count and ServerCnx can redirect the request.
*/
@Test
public void testSubscribeOnMigratedTopicSkipsAddingConsumer() throws Exception {
@@ -155,7 +174,7 @@ public class NonPersistentTopicTest extends BrokerTestBase {
(NonPersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).get();
// Mark the local cluster as migrated so AbstractTopic.getMigratedClusterUrlAsync() (used by
- // Consumer.checkAndApplyTopicMigrationAsync()) resolves to a present migrated-cluster URL.
+ // the subscribe migration check) resolves to a present migrated-cluster URL.
ClusterUrl migratedUrl = new ClusterUrl("http://migrated:8080", "https://migrated:8443",
"pulsar://migrated:6650", "pulsar+ssl://migrated:6651");
admin.clusters().updateClusterMigration(conf.getClusterName(), true, migratedUrl);
@@ -200,23 +219,115 @@ public class NonPersistentTopicTest extends BrokerTestBase {
long usageBefore = spyTopic.currentUsageCount();
- org.apache.pulsar.broker.service.Consumer consumer =
- spyTopic.subscribe(option).get(10, TimeUnit.SECONDS);
- assertNotNull(consumer);
+ assertThatThrownBy(() -> spyTopic.subscribe(option).get(10, TimeUnit.SECONDS))
+ .hasCauseInstanceOf(TopicMigratedException.class);
- // The migration redirect must have been sent to the client.
- Mockito.verify(commandSender).sendTopicMigrated(Mockito.any(), Mockito.eq(1L),
- Mockito.eq(migratedUrl.getBrokerServiceUrl()), Mockito.eq(migratedUrl.getBrokerServiceUrlTls()));
+ // ServerCnx handles the failed subscribe and sends the redirect; the topic must not disconnect a consumer.
+ Mockito.verify(commandSender, Mockito.never()).sendTopicMigrated(Mockito.any(), Mockito.anyLong(),
+ Mockito.any(), Mockito.any());
// The core regression assertion: a migrated topic must never attach the consumer to the
// subscription. The buggy code added it before the async redirect/disconnect could run.
Mockito.verify(spySubscription, Mockito.never()).addConsumer(Mockito.any());
assertTrue(spySubscription.getConsumers().isEmpty());
- // No usage-count leak: handleConsumerAdded's increment is balanced by the disconnect's removeConsumer.
+ // The subscribe failure must balance handleConsumerAdded's increment without dispatcher removal.
assertEquals(spyTopic.currentUsageCount(), usageBefore);
}
+ @DataProvider
+ public Object[][] migrationSubscriptionTypes() {
+ return new Object[][] {
+ {SubscriptionType.Exclusive}, {SubscriptionType.Failover},
+ {SubscriptionType.Shared}, {SubscriptionType.Key_Shared}
+ };
+ }
+
+ @Test(dataProvider = "migrationSubscriptionTypes")
+ public void testMigratedSubscribeReleasesUsageCount(SubscriptionType subscriptionType) throws Exception {
+ String topicName = "non-persistent://prop/ns-abc/migration-cleanup-" + UUID.randomUUID();
+ String subName = "migration-sub";
+ @Cleanup
+ Consumer<byte[]> existingConsumer = pulsarClient.newConsumer().topic(topicName)
+ .subscriptionName(subName).subscriptionType(subscriptionType).subscribe();
+ NonPersistentTopic topic =
+ (NonPersistentTopic) pulsar.getBrokerService().getTopicReference(topicName).orElseThrow();
+ NonPersistentSubscription subscription = topic.getSubscriptions().get(subName);
+ BrokerService brokerService = pulsar.getBrokerService();
+ PulsarResources resources = Mockito.spy(pulsar.getPulsarResources());
+ Mockito.doReturn(resources).when(pulsar).getPulsarResources();
+ assertEquals(topic.currentUsageCount(), 1L);
+
+ // Let the new client finish lookup, but delay its subscribe ownership check until migration starts.
+ CompletableFuture<Void> resumeSubscribe = new CompletableFuture<>();
+ CountDownLatch subscribeStarted = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ CompletableFuture<Void> ownership = (CompletableFuture<Void>) invocation.callRealMethod();
+ return ownership.thenCompose(__ -> {
+ subscribeStarted.countDown();
+ return resumeSubscribe;
+ });
+ }).when(brokerService).checkTopicNsOwnership(topicName);
+ // Observe only the pending subscribe's client; the existing consumer also receives a migration notification.
+ CompletableFuture<CommandTopicMigrated> migrationRedirect = new CompletableFuture<>();
+ ClientBuilderImpl clientBuilder = (ClientBuilderImpl) PulsarClient.builder()
+ .serviceUrl(pulsar.getBrokerServiceUrl());
+ @Cleanup
+ PulsarClient reconnectingClient = InjectedClientCnxClientBuilder.create(clientBuilder,
+ (clientConf, eventLoopGroup) -> new ClientCnx(InstrumentProvider.NOOP, clientConf, eventLoopGroup) {
+ @Override
+ protected void handleTopicMigrated(CommandTopicMigrated command) {
+ super.handleTopicMigrated(command);
+ migrationRedirect.complete(new CommandTopicMigrated().copyFrom(command));
+ }
+ });
+ reconnectingClient.newConsumer().topic(topicName).subscriptionName(subName)
+ .subscriptionType(subscriptionType).subscribeAsync();
+ assertTrue(subscribeStarted.await(10, TimeUnit.SECONDS));
+
+ ClusterUrl migratedUrl = new ClusterUrl("http://migrated:8080", null, "pulsar://migrated:6650", null);
+ admin.clusters().updateClusterMigration(conf.getClusterName(), true, migratedUrl);
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(() ->
+ AbstractTopic.getMigratedClusterUrlAsync(pulsar, topicName).get().isPresent());
+ CompletableFuture<Void> migration;
+ try {
+ // Keep the existing dispatcher in the subscription while the new subscribe observes migration.
+ // The migration flag is set before checkClusterMigration disconnects the existing consumer.
+ synchronized (subscription.getDispatcher()) {
+ migration = CompletableFuture.completedFuture(null)
+ .thenComposeAsync(__ -> topic.checkClusterMigration(), pulsar.getExecutor());
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).until(topic::isMigrated);
+ CountDownLatch migrationPolicyRead = new CountDownLatch(1);
+ Mockito.doAnswer(invocation -> {
+ migrationPolicyRead.countDown();
+ return invocation.callRealMethod();
+ }).when(resources).getClusterResources();
+ pulsar.getExecutor().execute(() -> resumeSubscribe.complete(null));
+ assertTrue(migrationPolicyRead.await(10, TimeUnit.SECONDS),
+ "The pending subscribe must reach its migration check with the existing subscription");
+ assertTrue(topic.getSubscriptions().get(subName) == subscription);
+ assertEquals(subscription.getConsumers().size(), 1,
+ "The old consumer must still be attached when the pending subscribe checks migration");
+ }
+ migration.get(10, TimeUnit.SECONDS);
+ Awaitility.await().atMost(10, TimeUnit.SECONDS).untilAsserted(() ->
+ assertEquals(topic.currentUsageCount(), 0L,
+ "Both the existing consumer and the redirected subscribe must release their usage"));
+ assertTrue(subscription.getConsumers().isEmpty());
+
+ CommandTopicMigrated redirect = migrationRedirect.get(10, TimeUnit.SECONDS);
+ assertEquals(redirect.getResourceType(), ResourceType.Consumer);
+ assertEquals(redirect.getBrokerServiceUrl(), migratedUrl.getBrokerServiceUrl());
+ assertFalse(redirect.hasBrokerServiceUrlTls());
+ admin.topics().delete(topicName, true);
+ } finally {
+ resumeSubscribe.complete(null);
+ Mockito.doCallRealMethod().when(brokerService).checkTopicNsOwnership(topicName);
+ Mockito.doCallRealMethod().when(resources).getClusterResources();
+ Mockito.doCallRealMethod().when(pulsar).getPulsarResources();
+ }
+ }
+
@Test
public void testSubscriptionsOnNonPersistentTopic() throws Exception {
final String topicName = "non-persistent://prop/ns-abc/topic_" + UUID.randomUUID();