[fix] [broker] Part-1: Replicator can not created successfully due to an orphan replicator in the previous topic owner

Author: poorbarcode

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

@@ -67,7 +67,16 @@ public abstract class AbstractReplicator {
     private volatile State state = State.Stopped;
 
     protected enum State {
-        Stopped, Starting, Started, Stopping
+        // The internal producer is stopped.
+        Stopped,
+        // Trying to create a new internal producer.
+        Starting,
+        // The internal producer has started, and tries copy data.
+        Started,
+        // The internal producer is trying to stop.
+        Stopping,
+        // The replicator is never used again. Pulsar will create a new Replicator when enable replication again.
+        Closed
     }
 
     public AbstractReplicator(String localCluster, Topic localTopic, String remoteCluster, String remoteTopicName,
@@ -124,8 +133,7 @@ public synchronized void startProducer() {
                         replicatorId, waitTimeMs / 1000.0);
             }
             // BackOff before retrying
-            brokerService.executor().schedule(this::checkTopicActiveAndRetryStartProducer, waitTimeMs,
-                    TimeUnit.MILLISECONDS);
+            scheduleCheckTopicActiveAndStartProducer(waitTimeMs);
             return;
         }
         State state = STATE_UPDATER.get(this);
@@ -150,10 +158,8 @@ public synchronized void startProducer() {
                 long waitTimeMs = backOff.next();
                 log.warn("[{}] Failed to create remote producer ({}), retrying in {} s",
                         replicatorId, ex.getMessage(), waitTimeMs / 1000.0);
-
                 // BackOff before retrying
-                brokerService.executor().schedule(this::checkTopicActiveAndRetryStartProducer, waitTimeMs,
-                        TimeUnit.MILLISECONDS);
+                scheduleCheckTopicActiveAndStartProducer(waitTimeMs);
             } else {
                 log.warn("[{}] Failed to create remote producer. Replicator state: {}", replicatorId,
                         STATE_UPDATER.get(this), ex);
@@ -163,16 +169,38 @@ public synchronized void startProducer() {
 
     }
 
-    protected void checkTopicActiveAndRetryStartProducer() {
-        isLocalTopicActive().thenAccept(isTopicActive -> {
-            if (isTopicActive) {
-                startProducer();
+    protected void scheduleCheckTopicActiveAndStartProducer(final long waitTimeMs) {
+        brokerService.executor().schedule(() -> {
+            if (state == State.Closed) {
+                return;
             }
-        }).exceptionally(ex -> {
-            log.warn("[{}] Stop retry to create producer due to topic load fail. Replicator state: {}", replicatorId,
-                    STATE_UPDATER.get(this), ex);
-            return null;
-        });
+            CompletableFuture<Optional<Topic>> topicFuture = brokerService.getTopics().get(localTopicName);
+            if (topicFuture == null) {
+                // Topic closed.
+                return;
+            }
+            topicFuture.thenAccept(optional -> {
+                if (optional.isEmpty()) {
+                    // Topic closed.
+                    return;
+                }
+                if (optional.get() != localTopic) {
+                    // Topic closed and created a new one, current replicator is outdated.
+                    return;
+                }
+                // TODO check isClosing or Deleting.
+                Replicator replicator = localTopic.getReplicators().get(replicatorId);
+                if (replicator != AbstractReplicator.this) {
+                    // Current replicator has been closed, and created a new one.
+                    return;
+                }
+                startProducer();
+            }).exceptionally(ex -> {
+                log.warn("[{}] [{}] Stop retry to create producer due to unknown error. Replicator state: {}",
+                        localTopicName, replicatorId, STATE_UPDATER.get(this), ex);
+                return null;
+            });
+        }, waitTimeMs, TimeUnit.MILLISECONDS);
     }
 
     protected CompletableFuture<Boolean> isLocalTopicActive() {
@@ -188,14 +216,14 @@ protected CompletableFuture<Boolean> isLocalTopicActive() {
         }, brokerService.executor());
     }
 
-    protected synchronized CompletableFuture<Void> closeProducerAsync() {
+    protected synchronized CompletableFuture<Void> closeAsync(boolean onlyCloseProducer) {
         if (producer == null) {
-            STATE_UPDATER.set(this, State.Stopped);
+            updateStatus(onlyCloseProducer);
             return CompletableFuture.completedFuture(null);
         }
         CompletableFuture<Void> future = producer.closeAsync();
         return future.thenRun(() -> {
-            STATE_UPDATER.set(this, State.Stopped);
+            updateStatus(onlyCloseProducer);
             this.producer = null;
             // deactivate further read
             disableReplicatorRead();
@@ -206,11 +234,21 @@ protected synchronized CompletableFuture<Void> closeProducerAsync() {
                             + " retrying again in {} s",
                     replicatorId, ex.getMessage(), waitTimeMs / 1000.0);
             // BackOff before retrying
-            brokerService.executor().schedule(this::closeProducerAsync, waitTimeMs, TimeUnit.MILLISECONDS);
+            brokerService.executor().schedule(() -> closeAsync(onlyCloseProducer),
+                    waitTimeMs, TimeUnit.MILLISECONDS);
             return null;
         });
     }
 
+    protected void updateStatus(boolean onlyCloseProducer) {
+        if (onlyCloseProducer) {
+            // Only close producer.
+            STATE_UPDATER.set(this, State.Stopped);
+        } else {
+            // Close replicator.
+            STATE_UPDATER.set(this, State.Closed);
+        }
+    }
 
     public CompletableFuture<Void> disconnect() {
         return disconnect(false);
@@ -239,7 +277,7 @@ public synchronized CompletableFuture<Void> disconnect(boolean failIfHasBacklog)
                     getReplicatorReadPosition(), getNumberOfEntriesInBacklog());
         }
 
-        return closeProducerAsync();
+        return closeAsync(true);
     }
 
     public CompletableFuture<Void> remove() {

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

@@ -79,7 +79,7 @@ protected void readEntries(Producer<byte[]> producer) {
                             + " Closing it. Replicator state: {}",
                     replicatorId, STATE_UPDATER.get(this));
             STATE_UPDATER.set(this, State.Stopping);
-            closeProducerAsync();
+            closeAsync(false);
             return;
         }
     }

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

@@ -155,7 +155,7 @@ protected void readEntries(Producer<byte[]> producer) {
                             + " Closing it. Replicator state: {}",
                     replicatorId, STATE_UPDATER.get(this));
             STATE_UPDATER.set(this, State.Stopping);
-            closeProducerAsync();
+            closeAsync(false);
         }
 
     }
@@ -437,7 +437,7 @@ public void readEntriesFailed(ManagedLedgerException exception, Object ctx) {
                             + " already deleted and cursor is already closed {}, ({})",
                     replicatorId, ctx, exception.getMessage(), exception);
             // replicator is already deleted and cursor is already closed so, producer should also be stopped
-            closeProducerAsync();
+            closeAsync(false);
             return;
         } else if (!(exception instanceof TooManyRequestsException)) {
             log.error("[{}] Error reading entries at {}. Retrying to read in {}s. ({})",
@@ -556,7 +556,7 @@ public void deleteFailed(ManagedLedgerException exception, Object ctx) {
             log.error("[{}] Asynchronous ack failure because replicator is already deleted and cursor is already"
                             + " closed {}, ({})", replicatorId, ctx, exception.getMessage(), exception);
             // replicator is already deleted and cursor is already closed so, producer should also be stopped
-            closeProducerAsync();
+            closeAsync(false);
             return;
         }
         if (ctx instanceof PositionImpl) {