From 5c5efa511b7614c9f81b0e69a1cad749504fe878 Mon Sep 17 00:00:00 2001 From: Praveen Kumar <122990319+Praveenkumar76@users.noreply.github.com> Date: Sun, 2 Aug 2026 16:53:11 +0530 Subject: [PATCH] [fix][broker] Reject replicated subscriptions on non-replicated topics --- .../admin/impl/PersistentTopicsBase.java | 8 ++++ .../service/persistent/PersistentTopic.java | 14 ++++++- .../broker/admin/PersistentTopicsTest.java | 24 +++++++++++ .../ReplicatedSubscriptionConfigTest.java | 40 +++++++++++++++++++ 4 files changed, 85 insertions(+), 1 deletion(-) diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java index ff4fef6eef9de..f5c43c2a846f1 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java @@ -5355,6 +5355,14 @@ private void internalSetReplicatedSubscriptionStatusForNonPartitionedTopic( } if (topic instanceof PersistentTopic && sub instanceof PersistentSubscription) { + if (enabled + && !((PersistentSubscription) sub).isReplicated() + && !((PersistentTopic) topic).isReplicatedSubscriptionAllowed()) { + asyncResponse.resume(new RestException(Status.PRECONDITION_FAILED, + "Replicated subscriptions require topic replication")); + return; + } + if (!((PersistentSubscription) sub).setReplicated(enabled)) { asyncResponse.resume( new RestException(Status.INTERNAL_SERVER_ERROR, diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java index 3c4a404c32c12..a856cf001875b 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java @@ -1150,6 +1150,14 @@ private CompletableFuture getDurableSubscription(String subscripti Boolean replicated, Map subscriptionProperties) { CompletableFuture subscriptionFuture = new CompletableFuture<>(); + PersistentSubscription existingSubscription = subscriptions.get(subscriptionName); + if (Boolean.TRUE.equals(replicated) + && !isReplicatedSubscriptionAllowed() + && (existingSubscription == null || !existingSubscription.isReplicated())) { + subscriptionFuture.completeExceptionally( + new NotAllowedException("Replicated subscriptions require topic replication")); + return subscriptionFuture; + } if (checkMaxSubscriptionsPerTopicExceed(subscriptionName)) { subscriptionFuture.completeExceptionally(new NotAllowedException( "Exceed the maximum number of subscriptions of the topic: " + topic)); @@ -4607,6 +4615,10 @@ public CompletableFuture addSchemaIfIdleOrCheckCompatible(SchemaData schem }); } + public boolean isReplicatedSubscriptionAllowed() { + return topicPolicies.getReplicationClusters().get().size() > 1; + } + public synchronized void checkReplicatedSubscriptionControllerState() { AtomicBoolean shouldBeEnabled = new AtomicBoolean(false); subscriptions.forEach((name, subscription) -> { @@ -4626,7 +4638,7 @@ private synchronized void checkReplicatedSubscriptionControllerState(boolean sho boolean isCurrentlyEnabled = replicatedSubscriptionsController.isPresent(); boolean isEnableReplicatedSubscriptions = brokerService.pulsar().getConfiguration().isEnableReplicatedSubscriptions(); - boolean replicationEnabled = this.topicPolicies.getReplicationClusters().get().size() > 1; + boolean replicationEnabled = isReplicatedSubscriptionAllowed(); if (shouldBeEnabled && !isCurrentlyEnabled && isEnableReplicatedSubscriptions && replicationEnabled) { log.info("Enabling replicated subscriptions controller"); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java index 1cf3c4abd3632..8312b0cb9f887 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/PersistentTopicsTest.java @@ -1420,6 +1420,30 @@ public void testSetReplicatedSubscriptionStatus() { Assert.assertEquals(responseCaptor.getValue().getStatus(), Response.Status.NO_CONTENT.getStatusCode()); } + @Test + public void testCannotEnableReplicatedSubscriptionWithoutTopicReplication() throws Exception { + String topicName = BrokerTestUtil.newUniqueName( + "persistent://" + testTenant + "/" + testNamespaceLocal + + "/replicated-subscription-without-replication"); + String subName = "sub"; + + admin.topics().createNonPartitionedTopic(topicName); + admin.topics().createSubscription(topicName, subName, MessageId.latest); + + PulsarAdminException.PreconditionFailedException exception = + Assert.expectThrows(PulsarAdminException.PreconditionFailedException.class, + () -> admin.topics().setReplicatedSubscriptionStatus(topicName, subName, true)); + + Assert.assertTrue(exception.getMessage() + .contains("Replicated subscriptions require topic replication")); + Map replicatedStatus = + admin.topics().getReplicatedSubscriptionStatus(topicName, subName); + Assert.assertFalse(replicatedStatus.getOrDefault(topicName, false)); + + // Disabling must remain valid even when topic replication is not configured. + admin.topics().setReplicatedSubscriptionStatus(topicName, subName, false); + } + @SuppressWarnings("deprecation") @Test public void testGetMessageById() throws Exception { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionConfigTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionConfigTest.java index 604326203e876..4aca78b173d0d 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionConfigTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionConfigTest.java @@ -20,13 +20,19 @@ import static org.testng.Assert.assertFalse; import static org.testng.Assert.assertTrue; +import java.util.List; +import java.util.Set; import lombok.Cleanup; import org.apache.pulsar.broker.BrokerTestUtil; import org.apache.pulsar.broker.testcontext.PulsarTestContext; import org.apache.pulsar.client.api.Consumer; import org.apache.pulsar.client.api.ProducerConsumerBase; +import org.apache.pulsar.client.api.PulsarClientException; import org.apache.pulsar.client.api.Schema; +import org.apache.pulsar.common.policies.data.ClusterData; +import org.apache.pulsar.common.policies.data.TenantInfoImpl; import org.apache.pulsar.common.policies.data.TopicStats; +import org.testng.Assert; import org.testng.annotations.AfterClass; import org.testng.annotations.BeforeClass; import org.testng.annotations.Test; @@ -34,11 +40,18 @@ @Test(groups = "broker") public class ReplicatedSubscriptionConfigTest extends ProducerConsumerBase { + private static final String REMOTE_CLUSTER = "remote"; + @Override @BeforeClass public void setup() throws Exception { super.internalSetup(); super.producerBaseSetup(); + + admin.clusters().createCluster(REMOTE_CLUSTER, + ClusterData.builder().serviceUrl(pulsar.getWebServiceAddress()).build()); + admin.tenants().updateTenant("public", + new TenantInfoImpl(Set.of("appid1", "appid2"), Set.of("test", REMOTE_CLUSTER))); } @Override @@ -53,10 +66,16 @@ protected void customizeMainPulsarTestContextBuilder(PulsarTestContext.Builder p pulsarTestContextBuilder.enableOpenTelemetry(true); } + private void enableTopicReplication(String topic) throws Exception { + admin.topics().createNonPartitionedTopic(topic); + admin.topics().setReplicationClusters(topic, List.of("test", REMOTE_CLUSTER)); + } + @Test public void createReplicatedSubscription() throws Exception { this.conf.setEnableReplicatedSubscriptions(true); String topic = BrokerTestUtil.newUniqueName("createReplicatedSubscription"); + enableTopicReplication(topic); @Cleanup Consumer consumer = pulsarClient.newConsumer(Schema.STRING) @@ -79,6 +98,7 @@ public void createReplicatedSubscription() throws Exception { public void upgradeToReplicatedSubscription() throws Exception { this.conf.setEnableReplicatedSubscriptions(true); String topic = BrokerTestUtil.newUniqueName("upgradeToReplicatedSubscription"); + enableTopicReplication(topic); Consumer consumer = pulsarClient.newConsumer(Schema.STRING) .topic(topic) @@ -105,6 +125,7 @@ public void upgradeToReplicatedSubscription() throws Exception { public void upgradeToReplicatedSubscriptionAfterRestart() throws Exception { this.conf.setEnableReplicatedSubscriptions(true); String topic = BrokerTestUtil.newUniqueName("upgradeToReplicatedSubscriptionAfterRestart"); + enableTopicReplication(topic); Consumer consumer = pulsarClient.newConsumer(Schema.STRING) .topic(topic) @@ -129,6 +150,25 @@ public void upgradeToReplicatedSubscriptionAfterRestart() throws Exception { consumer.close(); } + @Test + public void rejectReplicatedSubscriptionWithoutTopicReplication() throws Exception { + this.conf.setEnableReplicatedSubscriptions(true); + String topic = BrokerTestUtil.newUniqueName( + "persistent://my-property/my-ns/rejectReplicatedSubscription"); + + PulsarClientException.NotAllowedException exception = + Assert.expectThrows(PulsarClientException.NotAllowedException.class, + () -> pulsarClient.newConsumer(Schema.STRING) + .topic(topic) + .subscriptionName("sub") + .replicateSubscriptionState(true) + .subscribe()); + + assertTrue(exception.getMessage() + .contains("Replicated subscriptions require topic replication")); + assertFalse(admin.topics().getStats(topic).getSubscriptions().containsKey("sub")); + } + @Test public void testDisableReplicatedSubscriptions() throws Exception { this.conf.setEnableReplicatedSubscriptions(false);