From b7b20bb6dfea9abf0b45285580a39691edf70757 Mon Sep 17 00:00:00 2001
From: technoboy <technoboy@apache.org>
Date: Wed, 10 Nov 2021 18:33:07 +0800
Subject: [PATCH] [pulsar-broker] Add stop replicator producer logic when start
 replicator cluster failed.

---
 .../apache/pulsar/broker/service/persistent/PersistentTopic.java | 1 +
 1 file changed, 1 insertion(+)

diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
index 22df2dfb28bcc..cdc0e0020aa52 100644
--- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
+++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java
@@ -280,6 +280,7 @@ public PersistentTopic(String topic, ManagedLedger ledger, BrokerService brokerS
                     log.warn("[{}] failed to start replication", topic, e);
                 }
                 if (!isReplicatorStarted) {
+                    stopReplProducers();
                     throw new NamingException(
                             PersistentTopic.this.getName() + " Failed to start replicator " + remoteCluster);
                 }