[pulsar-broker] Add stop replicator producer logic when start replicator cluster failed

[Technoboy-]

Motivation

When create persistent topic in BrokerService#createPersistentTopic:

public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
                            public void openLedgerComplete(ManagedLedger ledger, Object ctx) {
                            try {
                                PersistentTopic persistentTopic = isSystemTopic(topic)
                                        ? new SystemTopic(topic, ledger, BrokerService.this)
                                        : new PersistentTopic(topic, ledger, BrokerService.this); // @ Here may throw NamingException
                                ...
                                }).thenRun(() -> {
                                    ...
                                }).exceptionally((ex) -> {
                                    log.warn(
                                            "Replication or dedup check failed."
                                                    + " Removing topic from topics list {}, {}",
                                            topic, ex);
                                    persistentTopic.stopReplProducers().whenComplete((v, exception) -> {
                                        topics.remove(topic, topicFuture);
                                        topicFuture.completeExceptionally(ex);
                                    });

                                    return null;
                                });
                            } catch (NamingException | PulsarServerException e) {
                                log.warn("Failed to create topic {}-{}", topic, e.getMessage());
                                pulsar.getExecutor().execute(() -> topics.remove(topic, topicFuture));
                                topicFuture.completeExceptionally(e);
                            }
                        }

It may throw NamingException when new PersistentTopic. We don't stop replicator producers in the NamingException catch block. This will result in some replicator producers being created and keep in the remote clusters producers list.
Any later retry to add replicator producers fails with the below logs :

Nov 05 13:25:31 pulsar-broker-node-1 pulsar[11435]: 13:25:31.054 [ForkJoinPool.commonPool-worker-29] ERROR org.apache.pulsar.broker.service.ServerCnx - [/10.10.0.47:34976] Failed to add producer to topic persistent://SD/operator20018/groupsettings: producerId=17690, org.apache.pulsar.broker.service.BrokerServiceException$NamingException: Producer with name 'pulsar.repl.ctp-pulsar-stage-cluster' is already connected to topic
Nov 05 13:25:33 pulsar-broker-node-1 pulsar[11435]: 13:25:33.351 [ForkJoinPool.commonPool-worker-29] ERROR org.apache.pulsar.broker.service.ServerCnx - [/10.10.0.47:34952] Failed to add producer to topic persistent://SD/operator20018/groupsettings: producerId=17694, org.apache.pulsar.broker.service.BrokerServiceException$NamingException: Producer with name 'pulsar.repl.ctp-pulsar-stage-cluster' is already connected to topic
Nov 05 13:25:33 pulsar-broker-node-1 pulsar[11435]: 13:25:33.670 [ForkJoinPool.commonPool-worker-29] ERROR org.apache.pulsar.broker.service.ServerCnx - [/10.10.0.47:34970] Failed to add producer to topic persistent://SD/operator20018/groupsettings: producerId=17696,

Documentation

• [ x ] no-need-doc

[lhotari]

@Technoboy- It seems that this solution doesn't work in all cases.

I've been debugging the issue and it looks like the changes in #11804 limited the producer replacement to one connection (&& Objects.equals(cnx, other.cnx)). The replacement won't happen if another connection is chosen on the client side for creating the next producer. The changes in #11804 were refactored further in #12846.
I'm not sure about the reason yet, just making some initial guesses.