[pulsar-broker] Add stop replicator producer logic when start replica…

…tor cluster failed. (apache#12724)

(cherry picked from commit fc7cba4)

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

@@ -269,6 +269,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);
                 }