[improve][broker] Fix replicated subscriptions race condition with mark delete update and snapshot completion

[lhotari]

Fixes #10054

Motivation

There has been a long-standing issue (#10054) where replicated subscriptions were not working correctly across clusters. The test case testReplicatedSubscriptionAcrossTwoRegions demonstrates this problem:

pulsar/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/ReplicatedSubscriptionTest.java

Lines 107 to 116 in a1a2b36

	public void testReplicatedSubscriptionAcrossTwoRegions() throws Exception {
	String namespace = BrokerTestUtil.newUniqueName("pulsar/replicatedsubscription");
	String topicName = "persistent://" + namespace + "/mytopic";
	String subscriptionName = "cluster-subscription";
	// Subscription replication produces duplicates, https://github.com/apache/pulsar/issues/10054
	// TODO: duplications shouldn't be allowed, change to "false" when fixing the issue
	boolean allowDuplicates = true;
	// this setting can be used to manually run the test with subscription replication disabled
	// it shows that subscription replication has no impact in behavior for this test case
	boolean replicateSubscriptionState = true;

Previously, this test would only pass when configured to allow duplicates (allowDuplicates = true), indicating that subscription replication was not functioning as expected.

This PR (#16651) addresses this issue:

Race Condition: A race condition existed between snapshot creation and mark-delete position updates, which could cause synchronization failures. If the mark delete position got updated before the snapshot was completed, it wouldn't be used even when it would have been suitable for the snapshot.

Modifications

• Handle ReplicatedSubscriptionsSnapshot on completion: When a snapshot is completed, it is now processed immediately, passing the current mark-delete position to trigger the update logic. This handles the case where the mark delete position was updated before the snapshot completed.

• Refactor snapshot handling logic: The previous snapshot handling code has been moved to notifyTheMarkDeletePositionChanged. Since snapshot handling is idempotent, it's safe to handle snapshots both when they arrive and when the mark-delete position moves forward.

Documentation

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

[github-actions]

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

[github-actions]

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]

Some assumptions aren't correct in this PR, will keep as draft until resolved.

I've validated the assumption and this PR makes sense. Before this PR a race condition is possible where the local cluster's mark delete position is updated before the snapshot is completed. That results in the case, that the most recent snapshot cannot immediately be used for sending the replicated subscription update back to the remote cluster (via writing the marker message to the topic).

[lhotari]

There's another related flaw in the replicated subscription snapshots that should be fixed. The commit 50ca143 is already in this direction, but a similar one is needed for the responses. Fixing this would also avoid the current solution of two rounds of snapshot requests when there are more than 2 remote clusters.
In addition, it's still required to have something to track the last non-marker message before the snapshot request. That position should be used instead of the snapshot request's message id as the position for the completed snapshot. That would mean reverting some of the reverted changes in commit dae88dc

[lhotari]

There's another related flaw in the replicated subscription snapshots that should be fixed. The commit 50ca143 is already in this direction, but a similar one is needed for the responses. Fixing this would also avoid the current solution of two rounds of snapshot requests when there are more than 2 remote clusters. In addition, it's still required to have something to track the last non-marker message before the snapshot request. That position should be used instead of the snapshot request's message id as the position for the completed snapshot. That would mean reverting some of the reverted changes in commit dae88dc

This turned out to be the incorrect direction. However, one change that continues to make sense is cb2d873. In the case of the replication snapshot response, the request position would be sent back instead of the latest topic position.

[lhotari]

This turned out to be the incorrect direction. However, one change that continues to make sense is cb2d873. In the case of the replication snapshot response, the request position would be sent back instead of the latest topic position.

I reverted the change related to this since it doesn't change the behavior when the last position is used. The response will get appended at the end and the subscription update will get sent from the other cluster only after all messages have been acknowledged before that point. For more than 2 cluster configurations, the 2 rounds of snapshots solution is used to avoid message loss (acknowledging messages that haven't been processed).