[fix][broker] Reject replicated subscriptions on non-replicated topics

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,

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentTopic.java

@@ -1150,6 +1150,14 @@ private CompletableFuture<Subscription> getDurableSubscription(String subscripti
                                                                    Boolean replicated,
                                                                    Map<String, String> subscriptionProperties) {
         CompletableFuture<Subscription> 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<Void> 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");

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<String, Boolean> 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 {

pulsar-broker/src/test/java/org/apache/pulsar/broker/service/persistent/ReplicatedSubscriptionConfigTest.java

@@ -20,25 +20,38 @@
 
 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;
 
 @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<String> 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<String> 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<String> 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);