[feat] [broker] PIP-188 support blue-green cluster migration [part-2]

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Replicator.java

@@ -49,4 +49,6 @@ default Optional<DispatchRateLimiter> getRateLimiter() {
     }
 
     boolean isConnected();
+
+    long getNumberOfEntriesInBacklog();
 }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/ServerCnx.java

@@ -1578,13 +1578,23 @@ private void buildProducerAndAddTopic(Topic topic, long producerId, String produ
             if (ex.getCause() instanceof BrokerServiceException.TopicMigratedException) {
                 Optional<ClusterUrl> clusterURL = getMigratedClusterUrl(service.getPulsar());
                 if (clusterURL.isPresent()) {
-                    log.info("[{}] redirect migrated producer to topic {}: producerId={}, {}", remoteAddress, topicName,
-                            producerId, ex.getCause().getMessage());
-                    commandSender.sendTopicMigrated(ResourceType.Producer, producerId,
-                            clusterURL.get().getBrokerServiceUrl(), clusterURL.get().getBrokerServiceUrlTls());
-                    closeProducer(producer);
-                    return null;
-
+                    if (topic.isReplicationBacklogExist()) {
+                        log.info("Topic {} is migrated but replication backlog exist: "
+                                        + "producerId = {}, producerName = {}, {}", topicName,
+                                producerId, producerName, ex.getCause().getMessage());
+                    } else {
+                        log.info("[{}] redirect migrated producer to topic {}: "
+                                        + "producerId={}, producerName = {}, {}", remoteAddress,
+                                topicName, producerId, producerName, ex.getCause().getMessage());
+                        boolean msgSent = commandSender.sendTopicMigrated(ResourceType.Producer, producerId,
+                                clusterURL.get().getBrokerServiceUrl(), clusterURL.get().getBrokerServiceUrlTls());
+                        if (!msgSent) {
+                            log.info("client doesn't support topic migration handling {}-{}-{}", topic,
+                                    remoteAddress, producerId);
+                        }
+                        closeProducer(producer);
+                        return null;
+                    }
                 } else {
                     log.warn("[{}] failed producer because migration url not configured topic {}: producerId={}, {}",
                             remoteAddress, topicName, producerId, ex.getCause().getMessage());

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/Topic.java

@@ -234,6 +234,8 @@ CompletableFuture<Subscription> createSubscription(String subscriptionName, Init
 
     boolean isBrokerPublishRateExceeded();
 
+    boolean isReplicationBacklogExist();
+
     void disableCnxAutoRead();
 
     void enableCnxAutoRead();

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentReplicator.java

@@ -248,7 +248,7 @@ protected Position getReplicatorReadPosition() {
     }
 
     @Override
-    protected long getNumberOfEntriesInBacklog() {
+    public long getNumberOfEntriesInBacklog() {
         // No-op
         return 0;
     }

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/nonpersistent/NonPersistentTopic.java

@@ -236,6 +236,11 @@ public void checkMessageDeduplicationInfo() {
         // No-op
     }
 
+    @Override
+    public boolean isReplicationBacklogExist() {
+        return false;
+    }
+
     @Override
     public void removeProducer(Producer producer) {
         checkArgument(producer.getTopic() == this);

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java

@@ -165,8 +165,8 @@ protected Position getReplicatorReadPosition() {
     }
 
     @Override
-    protected long getNumberOfEntriesInBacklog() {
-        return cursor.getNumberOfEntriesInBacklog(false);
+    public long getNumberOfEntriesInBacklog() {
+        return cursor.getNumberOfEntriesInBacklog(true);
     }
 
     @Override

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -575,6 +575,7 @@ public void addComplete(Position pos, ByteBuf entryData, Object ctx) {
 
     @Override
     public synchronized void addFailed(ManagedLedgerException exception, Object ctx) {
+        PublishContext callback = (PublishContext) ctx;
         if (exception instanceof ManagedLedgerFencedException) {
             // If the managed ledger has been fenced, we cannot continue using it. We need to close and reopen
             close();
@@ -587,7 +588,11 @@ public synchronized void addFailed(ManagedLedgerException exception, Object ctx)
                 List<CompletableFuture<Void>> futures = new ArrayList<>();
                 // send migration url metadata to producers before disconnecting them
                 if (isMigrated()) {
-                    producers.forEach((__, producer) -> producer.topicMigrated(getMigratedClusterUrl()));
+                    if (isReplicationBacklogExist()) {
+                        log.info("Topic {} is migrated but replication backlog exists. Closing producers.", topic);
+                    } else {
+                        producers.forEach((__, producer) -> producer.topicMigrated(getMigratedClusterUrl()));
+                    }
                 }
                 producers.forEach((__, producer) -> futures.add(producer.disconnect()));
                 disconnectProducersFuture = FutureUtil.waitForAll(futures);
@@ -599,8 +604,6 @@ public synchronized void addFailed(ManagedLedgerException exception, Object ctx)
                 return null;
             });
 
-            PublishContext callback = (PublishContext) ctx;
-
             if (exception instanceof ManagedLedgerAlreadyClosedException) {
                 if (log.isDebugEnabled()) {
                     log.debug("[{}] Failed to persist msg in store: {}", topic, exception.getMessage());
@@ -2510,6 +2513,18 @@ public CompletableFuture<Void> checkClusterMigration() {
         }
     }
 
+    public boolean isReplicationBacklogExist() {
+        ConcurrentOpenHashMap<String, Replicator> replicators = getReplicators();
+        if (replicators != null) {
+            for (Replicator replicator : replicators.values()) {
+                if (replicator.getNumberOfEntriesInBacklog() != 0) {
+                    return true;
+                }
+            }
+        }
+        return false;
+    }
+
     @Override
     public void checkGC() {
         if (!isDeleteWhileInactive()) {

pulsar-broker/src/test/java/org/apache/pulsar/broker/auth/MockedPulsarServiceBaseTest.java

@@ -112,14 +112,20 @@ public abstract class MockedPulsarServiceBaseTest extends TestRetrySupport {
     protected URI lookupUrl;
 
     protected boolean isTcpLookup = false;
-    protected static final String configClusterName = "test";
+    protected String configClusterName = "test";
 
     protected boolean enableBrokerInterceptor = false;
 
     public MockedPulsarServiceBaseTest() {
         resetConfig();
     }
 
+    protected void setupWithClusterName(String clusterName) throws Exception {
+        this.conf.setClusterName(clusterName);
+        this.configClusterName = clusterName;
+        this.internalSetup();
+    }
+
     protected PulsarService getPulsar() {
         return pulsar;
     }