[fix][broker] Ensure existing subscriptions use the initial replication policy

[nodece]

Motivation

The subscription replication will be changed if the org.apache.pulsar.client.api.ConsumerBuilder#replicateSubscriptionState is true:

1. New a consumer with replicateSubscriptionState is false, indicating that doesn't replicate this subscription.
2. Close the first consumer, and new a consumer with replicateSubscriptionState is true, the subscription will be replication.

Because we already store this state with the cursor properties, the existing subscription should use the initial replication policy, this configuration should only affect new subscriptions. Otherwise, the cursor properties are meaningless.

Modifications

• When new the org.apache.pulsar.broker.service.persistent.PersistentSubscription, which will call the org.apache.pulsar.broker.service.persistent.PersistentSubscription#setReplicated, so we can remove flip the subscription state logic.
• Update replicateSubscriptionState javadoc.

Documentation

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

[lhotari]

Currently, the subscription replication will be changed by the org.apache.pulsar.client.api.ConsumerBuilder#replicateSubscriptionState. The existing subscription should use the original replication policy, this configuration should only affect new subscriptions.

@nodece Has the behavior always been like this? Just wondering if this is a recent change which would be considered a regression? If not, changing the behavior could be problematic if there are users that rely on this existing behavior.

[nodece]

@lhotari The PR description has been updated.

@nodece Has the behavior always been like this?

Yes, it is always this.

Just wondering if this is a recent change which would be considered a regression?

Not.

If not, changing the behavior could be problematic if there are users that rely on this existing behavior.

Usually, the users don't switch the replicateSubscriptionState value. BTW, we persist the replication state to the cursor properties, when the subscription reloads, this state is still correct.

One notice that we should consider the false case, and persist.

[lhotari]

Perhaps there should be a warn log for cases where subscription time replicated flag coming from the client side at subscription (== consumer creation time) differs from the persisted state for replication?

[nodece]

Perhaps there should be a warn log for cases where subscription time replicated flag coming from the client side at subscription (== consumer creation time) differs from the persisted state for replication?

@lhotari This is a good idea. But ultimately, which configuration do we use?

[lhotari]

@lhotari This is a good idea. But ultimately, which configuration do we use?

@nodece What do you mean exactly? Providing an example would be helpful.

[nodece]

@lhotari This is a good idea. But ultimately, which configuration do we use?

@nodece What do you mean exactly? Providing an example would be helpful.

@lhotari

• case1: The persisted state is true, and the consumer config is false, which is the ultimate value?
• case2: The persisted state is false, and the consumer config is true, which is the ultimate value?

[lhotari]

@lhotari This is a good idea. But ultimately, which configuration do we use?

@nodece What do you mean exactly? Providing an example would be helpful.

@lhotari

• case1: The persisted state is true, and the consumer config is false, which is the ultimate value?
• case2: The persisted state is false, and the consumer config is true, which is the ultimate value?

@nodece Wouldn't the goal be to use the persisted state? There's currently no javadoc for replicateSubscriptionState in ConsumerBuilder:

pulsar/pulsar-client-api/src/main/java/org/apache/pulsar/client/api/ConsumerBuilder.java

Lines 433 to 437 in 04e89fe

	/**
	*
	* @param replicateSubscriptionState
	*/
	ConsumerBuilder<T> replicateSubscriptionState(boolean replicateSubscriptionState);

.
In the javadoc, it could be stated explicitly that it's not possible to change the replicateSubscriptionState by a client for a subscription after it has been created. Perhaps there should be a separate admin operation to change this later?
In any case, I think that documenting this change in a new PIP would also be useful for the future.
The original PIP-31 doesn't clearly define how replicateSubscriptionState should behave.

[lhotari]

I think that switching to use the word "initial" instead of "original" in the title and description of this PR could improve the clarity. The reason why I think it's more meaningful is due to the concept "initialSubscriptionPosition" that there is.

It would have been a better choice to call the method in the builder initialSubscriptionReplicationEnabled etc., however we don't want to break the API at this point and we'll just have to live with replicateSubscriptionState method name in the builder.

[nodece]

Wouldn't the goal be to use the persisted state?

@lhotari The goal is to use the persisted state, which is correct.

Perhaps there should be a separate admin operation to change this later?

org.apache.pulsar.client.admin.Topics#setReplicatedSubscriptionStatus can change this behavior.

This PR has been updated, could you have a chance to review this PR?