[improve][pip] PIP-398: Subscription replication on the broker, namespace and topic levels

pip/pip-398.md

@@ -0,0 +1,278 @@
+# PIP-398: Subscription replication on the broker, namespace and topic levels
+
+# Background knowledge
+
+https://github.com/apache/pulsar/pull/4299 introduces the subscription replication feature on the consumer level:
+
+```java
+Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("topic").replicateSubscriptionState(true /* false */)
+        .subscriptionName("sub").subscribe();
+```
+
+While this provides flexibility, it introduces overhead in managing replication for a large number of consumers. Users
+need to manually enable the `replicateSubscriptionState` flag for each consumer, which can become cumbersome in
+large-scale deployments.
+
+# Motivation
+
+The key motivation behind this PIP is to simplify subscription replication configuration, especially in failover
+scenarios. When a main cluster goes down and a backup cluster is activated, ensuring that subscription states are
+consistently replicated across clusters is critical for failover scenarios. By extending the replication configuration
+to the broker, namespace and topic levels, the system reduces the need for explicit consumer-level configuration.
+
+# Goals
+
+## In Scope
+
+- Provide management of subscription replication at the broker, namespace and topic levels using the Pulsar Admin CLI
+  and API.
+- Define the priority order for subscription replication configuration across different levels.
+
+## Out of Scope
+
+- Changes to the existing consumer-level `replicateSubscriptionState` configuration.
+- Changes to the subscription replication mechanism itself (how states are replicated across clusters).
+
+# High Level Design
+
+This PIP introduces the `replicateSubscriptionState` configuration at the **broker, namespace, and topic levels** to
+simplify the management of subscription replication.
+
+With this proposal, administrators can configure subscription replication at higher levels so that consumers
+automatically inherit the effective configuration.
+
+## Subscription Replication Evaluation
+
+Subscription replication policies follow Pulsar's hierarchical configuration model:
+
+```
+Topic > Namespace > Broker
+```
+
+The **highest level with a non-null value** determines the effective policy; lower levels are ignored.
+
+* **Broker, Namespace, and Topic levels** follow **enable-only semantics**:
+    * `true` → enables subscription replication.
+    * `false` → does **not disable** existing replicated subscriptions.
+    * `null` → not configured; continue to the next lower level.
+
+* **Consumer configuration** participates in the evaluation but is **not part of the policy hierarchy**:
+    * `true` → enables replication.
+    * `false` or `null` → does **not disable** replication; inherits the effective configuration from higher-level
+      policies.
+
+**Evaluation flow for each subscription:**
+
+1. Traverse `Topic → Namespace → Broker` to find the first non-null policy.
+2. If the effective policy is `true`, mark the subscription as replicated.
+3. Otherwise, evaluate the consumer configuration:
+    * `true` → enables replication.
+    * `false` or `null` → no change; existing replicated state is preserved.
+4. Explicitly **disabling a subscription** is only possible via **Pulsar Admin**, and succeeds **only if the effective policy is not `true`**.  
+   If a higher-level policy later becomes `true` (topic reload, namespace/broker update), the broker re-applies the effective policy and marks the subscription as replicated again.
+
+### Design Consistency with Consumer-level Logic
+
+This design builds upon the existing **consumer-level logic**, which already follows **enable-only semantics**:
+
+```java
+// https://github.com/apache/pulsar/blob/8798a4662300dfa3d330dbb0277073874cc2d86e/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java#L1153-L1156
+if (replicated != null && replicated && !subscription.isReplicated()) {
+    subscription.setReplicated(true);
+}
+```
+
+Key points:
+
+* **Consumer-level default changed** from `false` to `null` ([PR #23757](https://github.com/apache/pulsar/pull/23757)).
+  * Combined with the multi-level policy evaluation, this ensures **compatibility across all client languages and versions**:
+      * Older clients defaulting to `false`
+      * Newer clients defaulting to `null`
+  * Both behave correctly with **broker, namespace, and topic-level policies**.
+
+# Detailed Design
+
+## Design & Implementation Details
+
+### Broker level
+
+Add the field `replicateSubscriptionState` to the `org.apache.pulsar.broker.ServiceConfiguration` class
+to control subscription replication at the broker level:
+
+```java
+public class ServiceConfiguration implements PulsarConfiguration {
+    @FieldContext(
+            category = CATEGORY_SERVER,
+            required = false,
+            dynamic = false,
+            doc = "The default value for replicating subscription state."
+    )
+    private Boolean replicateSubscriptionState;
+}
+```
+
+### Namespace level
+
+1. Add the field `replicate_subscription_state` to the `org.apache.pulsar.common.policies.data.Policies` class
+   to control subscription replication at the namespace level:
+    ```java
+    public class Policies {
+        @SuppressWarnings("checkstyle:MemberName")
+        public Boolean replicate_subscription_state;
+    }
+    ```
+2. Add the management methods to the `org.apache.pulsar.client.admin.Namespaces` interface:
+     ```java
+     public interface Namespaces {
+         void setReplicateSubscriptionState(String namespace, boolean enabled) throws PulsarAdminException;
+         CompletableFuture<Void> setReplicateSubscriptionStateAsync(String namespace, boolean enabled);
+         Boolean getReplicateSubscriptionState(String namespace) throws PulsarAdminException;
+         CompletableFuture<Boolean> getReplicateSubscriptionStateAsync(String namespace);
+         void removeReplicateSubscriptionState(String namespace) throws PulsarAdminException;
+         CompletableFuture<Void> removeReplicateSubscriptionStateAsync(String namespace);
+     }
+     ```
+3. Implement the management methods in the `org.apache.pulsar.client.admin.internal.NamespacesImpl` class.
+
+### Topic level
+
+1. Add the field `Boolean replicateSubscriptionState` to the `org.apache.pulsar.common.policies.data.TopicPolicies`
+   class to enable subscription replication at the topic level:
+   ```java
+    public class TopicPolicies {
+         public Boolean replicateSubscriptionState;
+    }
+    ```
+2. Add the management methods to the `org.apache.pulsar.client.admin.TopicPolicies` interface:
+   ```java
+       public interface TopicPolicies {
+           void setReplicateSubscriptionState(String topic, boolean enabled) throws PulsarAdminException;
+           CompletableFuture<Void> setReplicateSubscriptionStateAsync(String topic, boolean enabled);
+           Boolean getReplicateSubscriptionState(String topic, boolean applied) throws PulsarAdminException;
+           CompletableFuture<Boolean> getReplicateSubscriptionStateAsync(String topic, boolean applied);
+           void removeReplicateSubscriptionState(String topic) throws PulsarAdminException;
+           CompletableFuture<Void> removeReplicateSubscriptionStateAsync(String topic);
+       }
+   ```
+3. Implement the management methods in the `org.apache.pulsar.client.admin.internal.TopicPoliciesImpl` class.
+
+### Consumer level
+
+No changes. When the consumer with `replicateSubscriptionState=true`, the subscription will be snapshot. If `false`, no
+operation will be performed.
+
+## Public-facing Changes
+
+### Public API
+
+> **Note:** Setting `false` at namespace or topic level is accepted but not applied (enable-only semantics). Use the
+`DELETE` endpoint or `remove-` command to clear the configuration instead.
+
+##### Namespace level
+
+- `POST /{tenant}/{namespace}/replicateSubscriptionState`: set the subscription replication configuration on the
+  namespace level.
+    - Content-Type: `application/json`
+    - Body: `true` or `false`.
+- `GET /{tenant}/{namespace}/replicateSubscriptionState`: get subscription replication configuration on the namespace
+  level.
+    - Response: `true`, `false`, or `null` (not configured)
+- `DELETE /{tenant}/{namespace}/replicateSubscriptionState`: remove the subscription replication configuration on the
+  namespace level.
+
+##### Topic level
+
+- `POST /{tenant}/{namespace}/{topic}/replicateSubscriptionState`: set the subscription replication configuration on
+  the topic level.
+    - Content-Type: `application/json`
+    - Body: `true` or `false`.
+- `GET /{tenant}/{namespace}/{topic}/replicateSubscriptionState`: get subscription replication configuration on the
+  topic level.
+    - Parameters:
+        - `applied=true`: get the effective configuration following `Topic > Namespace > Broker` precedence.
+        - `applied=false`: get the topic level configuration only.
+    - Response: `true`, `false`, or `null` (not configured)
+- `DELETE /{tenant}/{namespace}/{topic}/replicateSubscriptionState`: remove the subscription replication configuration
+  on the topic level.
+
+### CLI
+
+#### Namespace level
+
+- `pulsar-admin namespaces set-replicate-subscription-state <tenant>/<namespace> --enabled true/false` to
+  set the subscription replication on the namespace level.
+- `pulsar-admin namespaces get-replicate-subscription-state <tenant>/<namespace>` to get the subscription
+  replication configuration on the namespace level.
+- `pulsar-admin namespaces remove-replicate-subscription-state <tenant>/<namespace>` to remove the subscription
+  replication configuration on the namespace level.
+
+#### Topic level
+
+- `pulsar-admin topicPolicies set-replicate-subscription-state <tenant>/<namespace>/<topic> --enabled true/false`
+  to set the subscription replication on the topic level.
+- `pulsar-admin topicPolicies get-replicate-subscription-state <tenant>/<namespace>/<topic>` to get the
+  subscription replication configuration on the topic level.
+- `pulsar-admin topicPolicies remove-replicate-subscription-state <tenant>/<namespace>/<topic>` to remove the
+  subscription replication configuration on the topic level.
+
+# Security Considerations
+
+Both write and read operations require the necessary permissions, which already exist in Pulsar.
+
+## Namespace level
+
+- Write the subscription replication configuration:
+    - Required: `PolicyName.REPLICATED_SUBSCRIPTION` with `WRITE` permission.
+- Read the subscription replication configuration:
+    - Required: `PolicyName.REPLICATED_SUBSCRIPTION` with `READ` permission.
+
+## Topic level
+
+- Write the subscription replication configuration
+    - Required: `TopicOperation.SET_REPLICATED_SUBSCRIPTION_STATUS` permission.
+- Read the subscription replication configuration:
+    - Required: `TopicOperation.GET_REPLICATED_SUBSCRIPTION_STATUS` permission.
+
+# Backward & Forward Compatibility
+
+## Upgrade
+
+None.
+
+## Downgrade / Rollback
+
+If the broker is downgraded, users need to ensure that the subscription replication configuration is reset to maintain
+the correct replication behavior via the pulsar-admin CLI or API:
+
+- CLI
+  ```shell
+  # enable
+  pulsar-admin topics set-replicated-subscription-status <tenant>/<namespace>/<topic> -s <subName> --enable
+  # disable
+  pulsar-admin topics set-replicated-subscription-status <tenant>/<namespace>/<topic> -s <subName> --disable
+  ```
+- API
+  ```java
+  // enable
+  admin.topics().setReplicatedSubscriptionStatus(topic, subName, true);
+  // disable
+  admin.topics().setReplicatedSubscriptionStatus(topic, subName, false);
+  ```
+
+## Pulsar Geo-Replication Upgrade & Downgrade/Rollback Considerations
+
+None.
+
+# Alternatives
+
+No alternatives considered. This proposal naturally extends the existing consumer-level configuration to higher levels,
+following Pulsar's established hierarchical policy model.
+
+# General Notes
+
+None.
+
+# Links
+
+* Mailing List discussion thread: https://lists.apache.org/thread/6tc51xwknypzl2q2d9rwr6z65ws5b0l0
+* Mailing List voting thread: https://lists.apache.org/thread/qh1y3qol52r2m98oqhv3z6y2yzvtdw82