[fix] [broker] Messages lost on the remote cluster when using topic level replication

[poorbarcode]

Motivation

Background

1. the method PersistentTopic.addReplicationCluster will remove orphan replication cursors, see [fix][broker] Ignore and remove the replicator cursor when the remote cluster is absent #19972
2. when a topic is loading up, it will initialize replicators by the cursors named pulsar.repl.<remote cluster>

Issue

• Mechanism 2 was called before topic-level policies initializing, so it will remove the replicator and cursor when a topic is loading up.
• After the topic level policies are initialized, the broker will create a new replicator with a new cursor(start at the latest position), and all the messages in backlog are lost.

This issue may cause messages to be lost when using topic-level Geo-Replication. For example:

• Enable topic level Geo-Replication
• Send messages on this topic
  • There are 3 ledgers under the topic: {L1: 0~50000, L2: 0~50000, L3: 0~5000}
  • The replication cursor came to L1: 10000 now.
• Unload the topic.
  • The Replication cursor was removed due to the remote cluster is not in the replicated cluster list(the topic level policies have not been loaded yet).
  • Create a new replication cursor after the topic-level policies are loaded. But the new cursor will start at the latest position(L3: 50001)
• The messages L1: 10001 ~ L3: 50000 were lost

Modifications

Fix the issue: make Mechanism 2 being called after the topic-level policies were loaded.

Documentation

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

Matching PR in forked repository

PR in forked repository: x

[lhotari]

I added some questions.

Since this issue is related to topic policies, I'd like to bring up to attention also issue #21303. Topic policy mutations aren't thread safe and just wondering if there are cases that could lead to replication clusters being removed if the policies get mutated in a way where replication clusters are lost. Do we have plans to address #21303?

[nodece]

Related to #22504, I think we need to wait for the ns and topic policies to load before loading the replicator.

BTW, the ns and topic policies are asynchronous, and we also need to use a single thread for serial updates. #21303 should be fixed first before we can handle GEO replication correctly.

[poorbarcode]

@nodece

BTW, the ns and topic policies are asynchronous, and we also need to use a single thread for serial updates. #21303 should be fixed first before we can handle GEO replication correctly.

Correct, that is what the current PR did