Fix for Infinite loop in PersistentReplicator.startProducer()

pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java

@@ -617,15 +617,20 @@ public synchronized CompletableFuture<Void> disconnect(boolean failIfHasBacklog)
             return disconnectFuture;
         }
 
+        if (STATE_UPDATER.get(this) == State.Stopping) {
+            // Do nothing since the all "STATE_UPDATER.set(this, Stopping)" instructions are followed by closeProducerAsync()
+            // which will at some point change the state to stopped
+            return CompletableFuture.completedFuture(null);
+        }
+
         if (producer != null && (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, cursor.getMarkDeletedPosition(), cursor.getNumberOfEntriesInBacklog());
             return closeProducerAsync();
-        } else {
-            // If there's already a reconnection happening, signal to close it whenever it's ready
-            STATE_UPDATER.set(this, State.Stopping);
         }
+
+        STATE_UPDATER.set(this, State.Stopped);
         return CompletableFuture.completedFuture(null);
     }
 

pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java

@@ -33,6 +33,7 @@
 import static org.testng.Assert.assertTrue;
 import static org.testng.Assert.fail;
 
+import java.lang.reflect.Field;
 import java.lang.reflect.Method;
 import java.net.InetSocketAddress;
 import java.net.URL;
@@ -67,8 +68,11 @@
 import org.mockito.Mockito;
 import org.mockito.invocation.InvocationOnMock;
 import org.mockito.stubbing.Answer;
+import org.mockito.verification.VerificationMode;
+import org.powermock.api.mockito.PowerMockito;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import org.testng.Assert;
 import org.testng.annotations.AfterMethod;
 import org.testng.annotations.BeforeMethod;
 import org.testng.annotations.Test;
@@ -83,7 +87,9 @@
 import com.yahoo.pulsar.broker.service.persistent.PersistentReplicator;
 import com.yahoo.pulsar.broker.service.persistent.PersistentSubscription;
 import com.yahoo.pulsar.broker.service.persistent.PersistentTopic;
+import com.yahoo.pulsar.client.api.ProducerConfiguration;
 import com.yahoo.pulsar.client.api.PulsarClient;
+import com.yahoo.pulsar.client.impl.PulsarClientImpl;
 import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSubscribe;
 import com.yahoo.pulsar.common.api.proto.PulsarApi.CommandSubscribe.SubType;
 import com.yahoo.pulsar.common.api.proto.PulsarApi.MessageMetadata;
@@ -863,7 +869,7 @@ public void testAtomicReplicationRemoval() throws Exception {
         PersistentTopic topic = new PersistentTopic(globalTopicName, ledgerMock, brokerService);
         String remoteReplicatorName = topic.replicatorPrefix + "." + remoteCluster;
         ConcurrentOpenHashMap<String, PersistentReplicator> replicatorMap = topic.getReplicators();
-        ;
+
         final URL brokerUrl = new URL(
                 "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort());
         PulsarClient client = PulsarClient.create(brokerUrl.toString());
@@ -895,4 +901,42 @@ public void testAtomicReplicationRemoval() throws Exception {
         DeleteCursorCallback callback = captor.getValue();
         callback.deleteCursorComplete(null);
     }
+
+    @Test
+    public void testClosingReplicationProducerTwice() throws Exception {
+        final String globalTopicName = "persistent://prop/global/ns/testClosingReplicationProducerTwice";
+        String localCluster = "local";
+        String remoteCluster = "remote";
+        final ManagedLedger ledgerMock = mock(ManagedLedger.class);
+        doNothing().when(ledgerMock).asyncDeleteCursor(anyObject(), anyObject(), anyObject());
+        doReturn(new ArrayList<Object>()).when(ledgerMock).getCursors();
+
+        PersistentTopic topic = new PersistentTopic(globalTopicName, ledgerMock, brokerService);
+        String remoteReplicatorName = topic.replicatorPrefix + "." + localCluster;
+
+        final URL brokerUrl = new URL(
+                "http://" + pulsar.getAdvertisedAddress() + ":" + pulsar.getConfiguration().getBrokerServicePort());
+        PulsarClient client =  spy( PulsarClient.create(brokerUrl.toString()) );
+        PulsarClientImpl clientImpl = (PulsarClientImpl) client;
+        Field conf = PersistentReplicator.class.getDeclaredField("producerConfiguration");
+        conf.setAccessible(true);
+
+        ManagedCursor cursor = mock(ManagedCursorImpl.class);
+        doReturn(remoteCluster).when(cursor).getName();
+        brokerService.getReplicationClients().put(remoteCluster, client);
+        PersistentReplicator replicator = new PersistentReplicator(topic, cursor, localCluster, remoteCluster, brokerService);
+
+        doReturn(new CompletableFuture<Producer>()).when(clientImpl).createProducerAsync(globalTopicName, (ProducerConfiguration) conf.get(replicator), remoteReplicatorName);
+
+        replicator.startProducer();
+        verify(clientImpl).createProducerAsync(globalTopicName, (ProducerConfiguration) conf.get(replicator), remoteReplicatorName);
+
+        replicator.disconnect(false);
+        replicator.disconnect(false);
+
+        replicator.startProducer();
+
+        verify(clientImpl, Mockito.times(2)).createProducerAsync(globalTopicName, (ProducerConfiguration) conf.get(replicator), remoteReplicatorName);       
+    }
+
 }