[fix][broker] Do not enable replicated subscriptions on scalable topic segments - #26679

Motivation

A segment:// topic of a scalable topic lives in a namespace whose replication_clusters apply to it. PersistentTopic.checkReplicatedSubscriptionControllerState enables a ReplicatedSubscriptionsController when a subscription is replicated, enableReplicatedSubscriptions is on and the topic has more than one replication cluster — and that last check looks at the namespace policy, which cannot tell a segment apart from a regular topic. So a V5 consumer created with replicateSubscriptionState(true) (forwarded to every per-segment v4 consumer) enabled the controller on each segment.

The segment DAG of a scalable topic is independent per cluster (PIP-460, Geo-Replication), so no remote cluster can ever answer the snapshot requests the controller writes into the segment; replicated subscriptions for scalable topics are explicitly deferred to a future sub-PIP.

Reproduced on master with the new test: the controller is present on the segment. On current master the classic replicator that a segment also starts never connects, which happens to trip the controller's "some cluster is not reachable" skip and keeps it from writing. A separate change stops segments from starting classic replicators; with that applied, a REPLICATED_SUBSCRIPTION_SNAPSHOT_REQUEST marker lands in the segment ledger every snapshot period. The two changes belong together.

Modifications

• Broker: checkReplicatedSubscriptionControllerState(boolean) — the single choke point for all enable paths (subscribe, topic load, namespace/topic policy update, received-marker force path) — returns early for a segment topic, with a DEBUG log. Only enabling is skipped; the disable branch still runs.
  The subscribe request is not rejected: the flag still arrives on the wire from older V5 milestones or direct subscribeSegmentAsync users, and a NotAllowedException would turn a harmless request into a failing consumer. The subscription is created and keeps its replicated flag; only the controller is withheld.
• V5 client API: replicateSubscriptionState(boolean) is removed from StreamConsumerBuilder and QueueConsumerBuilder (and their implementations) until replicated subscriptions for scalable topics are designed. The method only shipped in the 5.0.0 milestones.
• CLI: -rs/--replicated moves from the base classes shared with the -v4 commands into consume-v4 (pulsar-client), consume-v4 and transaction-v4 (pulsar-perf) — the only commands that can still honour it. The V5 commands no longer accept it.

Verifying this change

This change added tests and can be verified as follows:

• ScalableTopicSegmentReplicatedSubscriptionTest (two-cluster OneWayReplicatorTestBase): creates a scalable topic in the replicated namespace, publishes to its segment and subscribes to the segment:// topic with replicateSubscriptionState=true the way a V5 consumer does, then asserts over several snapshot periods that the segment has no replicated subscriptions controller and no marker messages in its ledger. Fails on unpatched master (Expecting an empty Optional but was containing value: ReplicatedSubscriptionsController).
• Existing ReplicatedSubscriptionsControllerTest and ReplicatedSubscriptionTest still pass for regular topics; TestCmdConsume and PerformanceConsumerArgsTest pass.

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

If the box was checked, please highlight the changes

• 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

StreamConsumerBuilder#replicateSubscriptionState and QueueConsumerBuilder#replicateSubscriptionState are removed from the V5 client API (pulsar-client-api-v5, milestone-only so far). pulsar-client consume and pulsar-perf consume / transaction no longer accept -rs/--replicated; the -v4 variants keep it.