[fix][broker] release orphan replicator after replicator is closed.

[hanmz]

Motivation

When the replicator.producer start failed the method does not exit the loop(checkTopicActiveAndRetryStartProducer) even if the replicator is disconnect. You can reproduce by the test testExitRetryStartProducerAfterReplicatorDisconnect.

Modifications

If the replicator is already disconnected, stop to retryStartProducer.

Documentation

• doc
• doc-required
• doc-not-needed
• doc-complete

Matching PR in forked repository

PR in forked repository: hanmz#2

[poorbarcode]

Once the stat is set to closed, it can no longer be recovered to no-closed. If the replicator needs to keep working, it shouldn't be set to closed, right?

[poorbarcode]

Normally this is the case. But here, we can call the method startProducer again after call the method disconnect.
In fact, this class can be closed and opened again.
In fact, this class can be closed and opened again.

So we should only set the stat of Replicator to closed when it is exactly is closed, right?

[hanmz]

Once the stat is set to closed, it can no longer be recovered to no-closed. If the replicator needs to keep working, it shouldn't be set to closed, right?

Normally this is the case. But here, we can call the method startProducer again after call the method disconnect.
In fact, this class can be closed and opened again.

Essentially, what we want to do here is to allow the retry thread (checkTopicActiveAndRetryStartProducer) to end normally when the replicator is closed.

[poorbarcode]

Once we add a stat, it must be meaningful, right?

[poorbarcode]

@codelipenghui @Technoboy- What do you think about this?

[poorbarcode]

In fact, this class can be closed and opened again.

There is already a state [Stopped, Starting, Started, Stopping] is working on this case.

[hanmz]

There is currently no clear indication that this replicator will not be used again after it is disconnect.

Mabey it is not very reasonable to use the close status identifier here, but it can work normally.

I think it is also reasonable to use some other markers as judgment conditions.

[poorbarcode]

Can you provide a definition for this new state isClosed? How is it different from State.Stopped when it is true?

[poorbarcode]

Add a context:

This PR is trying to fix the issue below

• Close replication task and Start producer of replicator were executed at the same time.
• Start producer of replicator runs after Close replication task
• If the producer started failed, retry; if the next start also failed, retry again....loop.

Once the producer creates success, the loop will be stopped. Explain: After the cursor was closed. The method readEntries | readMoreEntries will get a CursorAlreadyClosedException, then the producer will be closed. see https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java#L440

if (exception instanceof CursorAlreadyClosedException) {
    log.error("[{}] Error reading entries because replicator is"
                    + " already deleted and cursor is already closed {}, ({})",
            replicatorId, ctx, exception.getMessage(), exception);
    // replicator is already deleted and cursor is already closed so, producer should also be stopped
    closeProducerAsync();
    return;
}

So this PR only trying to fix the scenario that the producer always creates fails.

[hanmz]

Add a context:

This PR is trying to fix the issue below

• Close replication task and Start producer of replicator were executed at the same time.
• Start producer of replicator runs after Close replication task
• If the producer started failed, retry; if the next start also failed, retry again....loop.

Once the producer creates success, the loop will be stopped. Explain: After the cursor was closed. The method readEntries | readMoreEntries will get a CursorAlreadyClosedException, then the producer will be closed. see https://github.com/apache/pulsar/blob/master/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentReplicator.java#L440

if (exception instanceof CursorAlreadyClosedException) {
    log.error("[{}] Error reading entries because replicator is"
                    + " already deleted and cursor is already closed {}, ({})",
            replicatorId, ctx, exception.getMessage(), exception);
    // replicator is already deleted and cursor is already closed so, producer should also be stopped
    closeProducerAsync();
    return;
}

So this PR only trying to fix the scenario that the producer always creates fails.

YES That's right

[Jason918]

@poorbarcode PTAL.

[Jason918]

@hanmz Please check the CI failure.

[hanmz]

/pulsarbot rerun-failure-checks

[poorbarcode]

Current PR is pending on this concern.

[poorbarcode]

I think this new state is not meaningful in the current design.

[poorbarcode]

#21946 will fix the issue that the current PR tries to fix.