diff --git a/pip/pip-398.md b/pip/pip-398.md new file mode 100644 index 0000000000000..199dc3c99bc15 --- /dev/null +++ b/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 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 setReplicateSubscriptionStateAsync(String namespace, boolean enabled); + Boolean getReplicateSubscriptionState(String namespace) throws PulsarAdminException; + CompletableFuture getReplicateSubscriptionStateAsync(String namespace); + void removeReplicateSubscriptionState(String namespace) throws PulsarAdminException; + CompletableFuture 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 setReplicateSubscriptionStateAsync(String topic, boolean enabled); + Boolean getReplicateSubscriptionState(String topic, boolean applied) throws PulsarAdminException; + CompletableFuture getReplicateSubscriptionStateAsync(String topic, boolean applied); + void removeReplicateSubscriptionState(String topic) throws PulsarAdminException; + CompletableFuture 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 / --enabled true/false` to + set the subscription replication on the namespace level. +- `pulsar-admin namespaces get-replicate-subscription-state /` to get the subscription + replication configuration on the namespace level. +- `pulsar-admin namespaces remove-replicate-subscription-state /` to remove the subscription + replication configuration on the namespace level. + +#### Topic level + +- `pulsar-admin topicPolicies set-replicate-subscription-state // --enabled true/false` + to set the subscription replication on the topic level. +- `pulsar-admin topicPolicies get-replicate-subscription-state //` to get the + subscription replication configuration on the topic level. +- `pulsar-admin topicPolicies remove-replicate-subscription-state //` 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 // -s --enable + # disable + pulsar-admin topics set-replicated-subscription-status // -s --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