[improve] [broker] Handle ReplicatedSubscriptionsSnapshots when it arrives to fix issues in replicating subscription state to remote clusters

[lhotari]

Fixes #10054

Motivation

There's a long-standing issue #10054 where replicated subscriptions were not working correctly across clusters. The test case testReplicatedSubscriptionAcrossTwoRegions demonstrate the issue:

https://github.com/apache/pulsar/blob/a1a2b363cfaa1bbc38933a742484a70a0a56e761/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatedSubscriptionTest.java#L107-L116

the test has been passing only when allowing duplicates (allowDuplicates = true in test case) in received messages, thus subscription replication is not working as expected.

Modifications

• Handle ReplicatedSubscriptionsSnapshot when it arrives, passing the current mark delete position to the update.
• move the previous logic to handle the snapshot to notifyTheMarkDeletePositionMoveForwardIfNeeded. Since the snapshot handling is idempotent, it's okay to handle it in both when the snapshot arrives or when the mark delete position moves forward.

Documentation

• [ ] doc
• [ ] doc-required
• [x] doc-not-needed
• [ ] doc-complete

[github-actions[bot]]

@lhotari Please provide a correct documentation label for your PR. Instructions see Pulsar Documentation Label Guide.

[github-actions[bot]]

The pr had no activity for 30 days, mark with Stale label.

[yangl]

@codelipenghui PTAL

[poorbarcode]

Since we will start the RC version of 3.0.0 on 2023-04-11, I will change the label/milestone of PR who have not been merged.

• The PR of type feature/improve is deferred to 3.1.0
• The PR of type fix is deferred to 3.0.1

So drag this PR to 3.0.1

[lhotari]

/pulsarbot rerun-failure-checks