[fix][broker] Reject replicated subscriptions on non-replicated topics - #26262

Fixes #24090

Motivation

Pulsar currently allows a replicated subscription to be created or enabled even when the topic itself is not replicated across multiple clusters.

In this situation, the subscription is marked as replicated on the local cluster, but there is no remote topic or subscription to receive the replicated subscription state. This creates a misleading configuration where replicated subscriptions appear enabled but cannot function.

Modifications

• Added validation to reject the creation of a new replicated subscription when the topic has fewer than two effective replication clusters.
• Added validation to reject upgrading an existing non-replicated subscription to a replicated subscription when topic replication is not enabled.
• Added the same validation to the admin API used by setReplicatedSubscriptionStatus.
• Continued to allow:
  • disabling replicated subscription status;
  • reconnecting to an existing subscription that is already marked as replicated.
• Reused the validation when determining whether the replicated-subscription controller should be enabled.
• Updated existing positive tests to explicitly configure topic replication across two clusters.

Verifying this change

• Make sure that the change passes the CI checks.

This change added tests and can be verified as follows:

• Added ReplicatedSubscriptionConfigTest.rejectReplicatedSubscriptionWithoutTopicReplication to verify that consumer-based replicated-subscription creation is rejected on a non-replicated topic.
• Added PersistentTopicsTest.testCannotEnableReplicatedSubscriptionWithoutTopicReplication to verify that the admin API rejects enabling replicated subscription status on a non-replicated topic.
• Verified that disabling replicated subscription status is still allowed.
• Verified existing replicated-subscription creation, upgrade, restart, broker-disabled, and admin API scenarios.

The following targeted tests passed locally:

./gradlew :pulsar-broker:test \
  --tests "org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionConfigTest.rejectReplicatedSubscriptionWithoutTopicReplication" \
  --tests "org.apache.pulsar.broker.admin.PersistentTopicsTest.testCannotEnableReplicatedSubscriptionWithoutTopicReplication" \
  -PskipJavaVersionCheck \
  -DtestRetryCount=0 \
  --no-daemon \
  --no-configuration-cache

./gradlew :pulsar-broker:test \
  --tests "org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionConfigTest.createReplicatedSubscription" \
  --tests "org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionConfigTest.upgradeToReplicatedSubscription" \
  --tests "org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionConfigTest.upgradeToReplicatedSubscriptionAfterRestart" \
  --tests "org.apache.pulsar.broker.service.persistent.ReplicatedSubscriptionConfigTest.testDisableReplicatedSubscriptions" \
  --tests "org.apache.pulsar.broker.admin.PersistentTopicsTest.testSetReplicatedSubscriptionStatus" \
  -PskipJavaVersionCheck \
  -DtestRetryCount=0 \
  --no-daemon \
  --no-configuration-cache

Does this pull request potentially affect one of the following parts:

• Dependencies (add or upgrade a dependency)
• The public API
• The schema
• The default values of configurations
• The threading model
• The binary protocol
• The REST endpoints
• The admin CLI options
• The metrics
• Anything that affects deployment

The replicated-subscription status REST endpoint now returns a precondition failure when enabling replication for a subscription whose topic is not replicated across multiple clusters.