From 6777c1672b79be576dd1b304b047973988943404 Mon Sep 17 00:00:00 2001 From: jai1 Date: Fri, 3 Mar 2017 16:48:05 -0800 Subject: [PATCH] Fix for Infinite loop in PersistentReplicator.startProducer() (#275) * Fix for Infinite loop in PersistentReplicator.startProducer() * Fix for Infinite loop in PersistentReplicator.startProducer() --- .../persistent/PersistentReplicator.java | 11 +++-- .../broker/service/PersistentTopicTest.java | 46 ++++++++++++++++++- 2 files changed, 53 insertions(+), 4 deletions(-) diff --git a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java index 8c92323434d33..f608a90e35d8d 100644 --- a/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java +++ b/pulsar-broker/src/main/java/com/yahoo/pulsar/broker/service/persistent/PersistentReplicator.java @@ -617,15 +617,20 @@ public synchronized CompletableFuture 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); } diff --git a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java b/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java index e8347fba9a4b7..edadd7b5004f9 100644 --- a/pulsar-broker/src/test/java/com/yahoo/pulsar/broker/service/PersistentTopicTest.java +++ b/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 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()).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()).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); + } + }