[Broker] Fix race condition in stopping replicator while it is starting

- there was a chance for a race condition that resulted in the replicator not disconnecting
  if disconnect was called when the replicator was starting.

upstream pull request apache#13412

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

@@ -190,15 +190,13 @@ public synchronized CompletableFuture<Void> disconnect(boolean failIfHasBacklog)
             return CompletableFuture.completedFuture(null);
         }
 
-        if (producer != null && (STATE_UPDATER.compareAndSet(this, State.Starting, State.Stopping)
-                || STATE_UPDATER.compareAndSet(this, State.Started, State.Stopping))) {
+        if (STATE_UPDATER.compareAndSet(this, State.Starting, State.Stopping)
+                || STATE_UPDATER.compareAndSet(this, State.Started, State.Stopping)) {
             log.info("[{}][{} -> {}] Disconnect replicator at position {} with backlog {}", topicName, localCluster,
                     remoteCluster, getReplicatorReadPosition(), getNumberOfEntriesInBacklog());
-            return closeProducerAsync();
         }
 
-        STATE_UPDATER.set(this, State.Stopped);
-        return CompletableFuture.completedFuture(null);
+        return closeProducerAsync();
     }
 
     public CompletableFuture<Void> remove() {