[fix][broker] Fix consumers are not redirected to migrated cluster (#…

…20928)
apache · Aug 18, 2023 · 4a9fec6 · 4a9fec6
1 parent 0cb1c78
commit 4a9fec6
Show file tree

Hide file tree

Showing 7 changed files with 64 additions and 2 deletions.
diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Consumer.java
@@ -818,6 +818,20 @@ public void topicMigrated(Optional<ClusterUrl> clusterUrl) {
         }
     }
 
+    public boolean checkAndApplyTopicMigration() {
+        if (subscription.isSubsciptionMigrated()) {
+            Optional<ClusterUrl> clusterUrl = AbstractTopic.getMigratedClusterUrl(cnx.getBrokerService().getPulsar());
+            if (clusterUrl.isPresent()) {
+                ClusterUrl url = clusterUrl.get();
+                cnx.getCommandSender().sendTopicMigrated(ResourceType.Consumer, consumerId, url.getBrokerServiceUrl(),
+                        url.getBrokerServiceUrlTls());
+                // disconnect consumer after sending migrated cluster url
+                disconnect();
+                return true;
+            }
+        }
+        return false;
+    }
     /**
      * Checks if consumer-blocking on unAckedMessages is allowed for below conditions:<br/>
      * a. consumer must have Shared-subscription<br/>

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java
@@ -1250,6 +1250,13 @@ protected void handleSubscribe(final CommandSubscribe subscribe) {
                                     });
                         })
                         .thenAccept(consumer -> {
+                            if (consumer.checkAndApplyTopicMigration()) {
+                                log.info("[{}] Disconnecting consumer {} on migrated subscription on topic {} / {}",
+                                        remoteAddress, consumerId, subscriptionName, topicName);
+                                consumers.remove(consumerId, consumerFuture);
+                                return;
+                            }
+
                             if (consumerFuture.complete(consumer)) {
                                 log.info("[{}] Created subscription on topic {} / {}",
                                         remoteAddress, topicName, subscriptionName);

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Subscription.java
@@ -102,6 +102,8 @@ default long getNumberOfEntriesDelayed() {
 
     CompletableFuture<Void> updateSubscriptionProperties(Map<String, String> subscriptionProperties);
 
+    boolean isSubsciptionMigrated();
+
     default void processReplicatedSubscriptionSnapshot(ReplicatedSubscriptionsSnapshot snapshot) {
         // Default is no-op
     }
@@ -130,4 +132,5 @@ static boolean isCumulativeAckMode(SubType subType) {
     static boolean isIndividualAckMode(SubType subType) {
         return SubType.Shared.equals(subType) || SubType.Key_Shared.equals(subType);
     }
+
 }
diff --git a/...c/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java b/...c/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentSubscription.java
@@ -271,6 +271,11 @@ public NonPersistentDispatcher getDispatcher() {
         return this.dispatcher;
     }
 
+    @Override
+    public boolean isSubsciptionMigrated() {
+        return topic.isMigrated();
+    }
+
     @Override
     public CompletableFuture<Void> close() {
         IS_FENCED_UPDATER.set(this, TRUE);

diff --git a/...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java b/...ker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentSubscription.java
@@ -1248,6 +1248,12 @@ void topicTerminated() {
         }
     }
 
+    @Override
+    public boolean isSubsciptionMigrated() {
+        log.info("backlog for {} - {}", topicName, cursor.getNumberOfEntriesInBacklog(true));
+        return topic.isMigrated() && cursor.getNumberOfEntriesInBacklog(true) <= 0;
+    }
+
     @Override
     public Map<String, String> getSubscriptionProperties() {
         return subscriptionProperties;

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -2576,8 +2576,14 @@ private boolean hasBacklogs() {
     public CompletableFuture<Void> checkClusterMigration() {
         Optional<ClusterUrl> clusterUrl = getMigratedClusterUrl();
         if (!isMigrated() && clusterUrl.isPresent()) {
-            log.info("{} triggering topic migration", topic);
-            return ledger.asyncMigrate().thenCompose(r -> null);
+            return ledger.asyncMigrate().thenApply(__ -> {
+                subscriptions.forEach((name, sub) -> {
+                    if (sub.isSubsciptionMigrated()) {
+                        sub.getConsumers().forEach(Consumer::checkAndApplyTopicMigration);
+                    }
+                });
+                return null;
+            });
         } else {
             return CompletableFuture.completedFuture(null);
         }

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ClusterMigrationTest.java
@@ -318,6 +318,27 @@ public void testClusterMigration(boolean persistent, SubscriptionType subType) t
         retryStrategically((test) -> topic2.getSubscription("s2") != null, 10, 500);
         assertFalse(topic2.getSubscription("s2").getConsumers().isEmpty());
 
+        // new sub on migration topic must be redirected immediately
+        Consumer<byte[]> consumerM = client1.newConsumer().topic(topicName).subscriptionType(subType)
+                .subscriptionName("sM").subscribe();
+        assertFalse(pulsar2.getBrokerService().getTopicReference(topicName).get().getSubscription("sM").getConsumers()
+                .isEmpty());
+        consumerM.close();
+
+        // migrate topic after creating subscription
+        String newTopicName = topicName + "-new";
+        consumerM = client1.newConsumer().topic(newTopicName).subscriptionType(subType)
+                .subscriptionName("sM").subscribe();
+        retryStrategically((t) -> pulsar1.getBrokerService().getTopicReference(newTopicName).isPresent(), 5, 100);
+        pulsar1.getBrokerService().getTopicReference(newTopicName).get().checkClusterMigration().get();
+        retryStrategically((t) ->
+        pulsar2.getBrokerService().getTopicReference(newTopicName).isPresent() &&
+        pulsar2.getBrokerService().getTopicReference(newTopicName).get().getSubscription("sM")
+                .getConsumers().isEmpty(), 5, 100);
+        assertFalse(pulsar2.getBrokerService().getTopicReference(newTopicName).get().getSubscription("sM").getConsumers()
+                .isEmpty());
+        consumerM.close();
+
         // publish messages to cluster-2 and consume them
         for (int i = 0; i < n; i++) {
             producer1.send("test2".getBytes());