[Replication] Fix race condition in stopping replicator while it is starting

[lhotari]

Motivation

• similar motivation as [pulsar-broker] Add stop replicator producer logic when start replicator cluster failed #12724

• there was a chance for a race condition that resulted in the replicator not disconnecting if disconnect was called when the replicator was starting.

Modifications

Modify the logic to move the chance for the race condition that leaves the replicator producer running

[github-actions]

@lhotari:Thanks for your contribution. For this PR, do we need to update docs?
(The PR template contains info about doc, which helps others know more about the changes. Can you provide doc-related info in this and future PR descriptions? Thanks)

[github-actions]

@lhotari:Thanks for providing doc info!

[eolivelli]

the code in closeProducerAsync is still not correct

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java

Line 160 in a5830ae

this.producer = null;

because we are setting "producer" to null in a different thread.

every access o the "producer" field should be inside a synchronized block

we could move producer to a AtomicReference variable, or in any case we have to rework how we are accessing that variable.

alternatively it is better to assign a non-null value only once and never set the value back to null again

[lhotari]

because we are setting "producer" to null in a different thread.

every access o the "producer" field should be inside a synchronized block

producer is a volatile field:

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractReplicator.java

Line 48 in a5830ae

protected volatile ProducerImpl producer;

I guess it's intentional that synchronized isn't used.
The goal of this PR was to fix the immediate issue which I believe gets fixed when the producer != null condition is removed.

[eolivelli]

@lhotari you are right

this patch makes sense and it fixes your problem.

I had never read the code for this part of Pulsar and I was surprised.

[michaeljmarshall]

LGTM.