Avoid triggering startReplProducer on newAddProducer as it may flips replicator state wrongly

[rdhabalia]

Motivation

On Race condition when replicationCluster in namespace-policy gets changed multiple times subsequently (repl-cluster added then removed) : replication producer couldn't close successfully.

1. New repl-cluster added: Adds Replicator in replicators and it tries to create Producer async.
2. Before Producer created repl-cluster deleted: It closes the cursor and flips replicator's state=stopped as Producer is not created yet and when producer will be created it will be closed if state is stopped
3. However, in between if any other producer gets created then it tries to start replicators again which flips replicator's state to starting and when producer will be created in step-2 it gets following exception:
   Error reading entries at null. Retrying to read in 67.938s. (Cursor was already closed) (7)

Modifications

Broker should not try to startReplicator again on newProducer-creation as it flips the state of replicator.

Result

It will avoid flipping up replicator's state wrongly and replicator's producer can be closed on replicator removal.

[merlimat]

@rdhabalia But we need to resume the replicator it was stopped. Replicator will be paused when no one is connected and there are no local subscriptions.

Now, whenever a new producer comes in, we need to restart the replicator. How would that work with this patch?

[rdhabalia]

But we need to resume the replicator it was stopped. Replicator will be paused when no one is connected and there are no local subscriptions.

Aren't we starting the replicator on topic loading. And due to inactivity when gc closes the replicator, it also deletes the topic. So, next time when producer/consumer will be connected, it forces topic to be loaded and that would also start the replicator. I think GC is the only usecase where broker pause the replicator.
Also, if remove-replicator signals producer to be closed after creation then we should avoid addProducer to overwrite the state. Another option could be check TopicAlreadyClosed on readFailure

[merlimat]

Aren't we starting the replicator on topic loading. And due to inactivity when gc closes the replicator, it also deletes the topic. So, next time when producer/consumer will be connected, it forces topic to be loaded and that would also start the replicator. I think GC is the only usecase where broker pause the replicator.

The replicator will be paused and the topic might not be deleted (eg: an incoming replicator is still connected and keeps the topic alive)

[merlimat]

Change lgtm. Should we have at least a test that adds and removes a cluster from the list multiple times to stress the race condition?

[rdhabalia]

Actually, this change will still create the same issue. if someone(addProducer) calls startProducer before producer is created then as state is Stopping, it will schedule startProducer. and as soon as producer will be closed, state will be flipped to Stopped so, startProducer will again try to create producer and try to readEntries using closed-cursor.

I think fundamental issue is: removeReplicator() disconnects producers and removes replicator from cache (replicators) asynchronously and it's not atomic. So, if someone(addProducer) tries to startReplProducer in between it then it will create inconsistency in behavior.

So, startReplProducer should read replicator-clusters directly from policies rather from cache(replicators)

I will add test-case if change looks fine.

[rdhabalia]

Added testcase to verify this race-condition.

[saandrews]

LGTM 👍 @merlimat Can you take a look?

[merlimat]

👍