[Bug] Replicated Subscriptions can be created on non-replicated topics

[shasank112001]

Search before asking

• I searched in the issues and found nothing similar.

Read release policy

• I understand that unsupported versions don't get bug fixes. I will attempt to reproduce the issue on a supported version of Pulsar client and Pulsar broker.

Version

OS- Rhel 9
Pulsar Version - 3.3.1
JDK Version - 17

Minimal reproduce step

1. Create a tenant and namespace with allowed clusters A and B.
2. Create a topic without enabling geo replication on namespace or topic level.
3. Connect a consumer on the topic with a replicated subscription on Cluster A.

What did you expect to see?

An error, as a replicated subscription should not be possible on a non-replicated topic.

What did you see instead?

No error, with the subscription created and working. If I use Pulsar-admin to get the replication status on Cluster A, the result is True.
If I switch to pulsar admin on Cluster B, the subscription doesn't exist.

Anything else?

No response

Are you willing to submit a PR?

• I'm willing to submit a PR!

[nodece]

Create a tenant and namespace with allowed clusters A and B.

It look like you enabled the GEO on the namespace.

[shasank112001]

@nodece. I did not. Namespaces have two fields in their policies, allowedClusters and replicatedClusters.
Setting replicatedClusters starts up namespace level replication.
We set allowed clusters as we wanted to perform topic level replication.

[nodece]

Could you share your data?

$ pulsar-admin topics get-replication-clusters <topic-name>
$ pulsar-admin topics get-replication-clusters <topic-name> --applied

[shasank112001]

The namespace Policies (The namespace was created on cluster A with no replication_clusters, but allowed_clusters = ["cluster-a", "cluster-b"] ->
{
"auth_policies": {
"namespace_auth": {},
"destination_auth": {},
"subscription_auth_roles": {}
},
"replication_clusters": [
"cluster-a"
],
"allowed_clusters": [
"cluster-a",
"cluster-b"
],
"bundles": {
"boundaries": [
"0x00000000",
"0x40000000",
"0x80000000",
"0xc0000000",
"0xffffffff"
],
"numBundles": 4
},
"backlog_quota_map": {},
"clusterDispatchRate": {},
"topicDispatchRate": {},
"subscriptionDispatchRate": {},
"replicatorDispatchRate": {},
"clusterSubscribeRate": {},
"publishMaxMessageRate": {},
"latency_stats_sample_rate": {},
"deleted": false,
"encryption_required": false,
"inactive_topic_policies": {
"inactiveTopicDeleteMode": "delete_when_no_subscriptions",
"maxInactiveDurationSeconds": 0,
"deleteWhileInactive": false
},
"subscription_auth_mode": "None",
"offload_threshold": -1,
"offload_threshold_in_seconds": -1,
"schema_compatibility_strategy": "UNDEFINED",
"is_allow_auto_update_schema": true,
"schema_validation_enforced": false,
"subscription_types_enabled": [],
"properties": {},
"migrated": false
}

The result of the commands you asked to run:

/pulsar/bin $ pulsar-admin topics get-replication-clusters persistent://tenant/namespace/topic
null
/pulsar/bin $ pulsar-admin topics get-replication-clusters persistent://tenant/namespace/topic --applied
[ "cluster-a" ]

[shasank112001]

Creating a subscription via the admin API and enabling replication on it via the api on cluster A. Then getting the replication status of the subscription on cluster A results in this output:
{
"persistent://tenant/namespace/topic": true
}

Getting the replication status of the subscription on cluster B results in this output:
{
"reason": "Subscription sub not found for topic persistent://tenant/namespace/topic"
}

[nodece]

This is normal. Once you set up the GEO in the topic, it will create the subscription on cluster B.

[shasank112001]

But I did not set up geo on the topic. The "cluster-a" on the namespace gets applied automatically when the namespace is created on cluster-a.
With allowed clusters = "cluster-a", "cluster-b" I assumed only the namespace metadata would be replicated. As subscriptions show up as replicated, but they are not getting replicated across the topics.

[nodece]

1. Please ensure replicateSubscriptionState value is false.

try (Consumer<byte[]> consumer = client.newConsumer()
        .topic(topicName)
        .subscriptionName(subscriptionName)
        .replicateSubscriptionState(false) // Please use false.
        .subscribe()) {
}

2. If you want to remove this config, please use the admin API. Note: when the consumer is restarting, this state will be reset, which depends on replicateSubscriptionState value.

# disable
bin/pulsar-admin topics set-replicated-subscription-status <topic-name> -s <sub-name> -d
# enable
bin/pulsar-admin topics set-replicated-subscription-status <topic-name> -s <sub-name> -e