[fix][broker] Allow empty replication clusters for namespace policies

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java

@@ -755,9 +755,6 @@ protected CompletableFuture<Void> internalSetNamespaceReplicationClusters(List<S
         return validateNamespacePolicyOperationAsync(namespaceName, PolicyName.REPLICATION, PolicyOperation.WRITE)
                 .thenCompose(__ -> validatePoliciesReadOnlyAccessAsync())
                 .thenApply(__ -> {
-                    if (CollectionUtils.isEmpty(clusterIds)) {
-                        throw new RestException(Status.PRECONDITION_FAILED, "ClusterIds should not be null or empty");
-                    }
                     if (!namespaceName.isGlobal() && !(clusterIds.size() == 1
                             && clusterIds.get(0).equals(pulsar().getConfiguration().getClusterName()))) {
                             throw new RestException(Status.PRECONDITION_FAILED,

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java

@@ -3346,10 +3346,6 @@ protected CompletableFuture<Void> internalSetBacklogQuota(BacklogQuota.BacklogQu
     }
 
     protected CompletableFuture<Void> internalSetReplicationClusters(List<String> clusterIds, boolean isGlobal) {
-        if (CollectionUtils.isEmpty(clusterIds)) {
-            return CompletableFuture.failedFuture(new RestException(Status.PRECONDITION_FAILED,
-                    "ClusterIds should not be null or empty"));
-        }
         Set<String> replicationClusters = Sets.newHashSet(clusterIds);
         return validatePoliciesReadOnlyAccessAsync()
                 .thenAccept(__ -> {

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java

@@ -43,6 +43,7 @@
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collection;
+import java.util.Collections;
 import java.util.EnumSet;
 import java.util.HashSet;
 import java.util.List;
@@ -696,6 +697,23 @@ public void testGlobalNamespaceReplicationConfiguration() throws Exception {
         resetBroker();
     }
 
+    @Test
+    public void testNamespaceEmptyReplicationClusters() throws Exception {
+        // Test setting empty replication clusters for global namespace
+        asyncRequests(rsp -> namespaces.setNamespaceReplicationClusters(rsp,
+                this.testGlobalNamespaces.get(0).getTenant(), 
+                this.testGlobalNamespaces.get(0).getCluster(),
+                this.testGlobalNamespaces.get(0).getLocalName(),
+                Collections.emptyList()));
+
+        // Verify empty replication clusters are set
+        Set<String> repClusters = (Set<String>) asyncRequests(rsp -> namespaces.getNamespaceReplicationClusters(rsp,
+                this.testGlobalNamespaces.get(0).getTenant(), 
+                this.testGlobalNamespaces.get(0).getCluster(),
+                this.testGlobalNamespaces.get(0).getLocalName()));
+        assertEquals(repClusters, Collections.emptySet());
+    }
+
     @Test
     public void testGetBundles() throws Exception {
         List<String> boundaries = List.of("0x00000000", "0x80000000", "0xffffffff");

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java

@@ -27,6 +27,7 @@
 import static org.testng.Assert.fail;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.List;
 import java.util.Objects;
 import java.util.Set;
@@ -102,7 +103,7 @@ public void setup() throws Exception {
         admin.clusters().createCluster("usw", ClusterData.builder().serviceUrl("http://broker-usw.com:8080").build());
         admin.clusters().createCluster("usc", ClusterData.builder().serviceUrl("http://broker-usc.com:8080").build());
         admin.tenants().createTenant(this.testTenant,
-                new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use", "usc", "usw")));
+                new TenantInfoImpl(Set.of("role1", "role2"), Set.of("use", "usc", "usw", "global")));
 
         createTestNamespaces(this.testLocalNamespaces);
 
@@ -236,4 +237,53 @@ public void testOperationDelayedDelivery() throws Exception {
             assertEquals(e.getResponse().getStatus(), Response.Status.PRECONDITION_FAILED.getStatusCode());
         }
     }
+
+    @Test
+    public void testNamespaceEmptyReplicationClusters() throws Exception {
+        String globalNamespaceName = "test-replication-ns";
+        asyncRequests(ctx -> namespaces.createNamespace(ctx, testTenant, globalNamespaceName, null));
+
+        // Test 1: Set empty replication clusters (core functionality)
+        asyncRequests(rsp -> namespaces.setNamespaceReplicationClusters(rsp, testTenant, 
+                globalNamespaceName, Collections.emptyList()));
+
+        // Verify empty replication clusters are set
+        assertEquals(asyncRequests(rsp -> namespaces.getNamespaceReplicationClusters(rsp,
+                testTenant, globalNamespaceName)), Collections.emptySet());
+
+        // Test 2: Set non-empty replication clusters and verify it works
+        asyncRequests(rsp -> namespaces.setNamespaceReplicationClusters(rsp, testTenant,
+                globalNamespaceName, List.of("use")));
+
+        assertEquals(asyncRequests(rsp -> namespaces.getNamespaceReplicationClusters(rsp,
+                testTenant, globalNamespaceName)), Set.of("use"));
+
+        // Test 3: Set empty replication clusters again to verify it works both ways  
+        asyncRequests(rsp -> namespaces.setNamespaceReplicationClusters(rsp, testTenant,
+                globalNamespaceName, Collections.emptyList()));
+
+        assertEquals(asyncRequests(rsp -> namespaces.getNamespaceReplicationClusters(rsp,
+                testTenant, globalNamespaceName)), Collections.emptySet());
+
+        // Clean up
+        String fullNamespaceName = testTenant + "/" + globalNamespaceName;
+        admin.namespaces().deleteNamespace(fullNamespaceName);
+    }
+
+    @Test
+    public void testNamespaceEmptyReplicationClustersNoAutoPopulation() throws Exception {
+        // Create a V2 namespace using admin client API
+        NamespaceName v2Namespace = NamespaceName.get(testTenant, "v2-test-ns");
+        admin.namespaces().createNamespace(v2Namespace.toString(), Set.of("use"));
+
+        // Set empty replication clusters
+        admin.namespaces().setNamespaceReplicationClusters(v2Namespace.toString(), Collections.emptySet());
+
+        // Verify it remains empty (should not auto-populate to local cluster)
+        List<String> repClusters = admin.namespaces().getNamespaceReplicationClusters(v2Namespace.toString());
+        assertEquals(repClusters, Collections.emptyList());
+
+        // Clean up
+        admin.namespaces().deleteNamespace(v2Namespace.toString());
+    }
 }

pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java

@@ -3505,8 +3505,11 @@ public void testReplicatorClusterApi() throws Exception {
         Awaitility.await().untilAsserted(()
                 -> assertNull(admin.topics().getReplicationClusters(topic, false)));
 
-        assertThrows(PulsarAdminException.PreconditionFailedException.class, () -> admin.topics()
-                .setReplicationClusters(topic, List.of()));
+        // Test that empty replication clusters are now allowed
+        admin.topics().setReplicationClusters(topic, Collections.emptyList());
+        Awaitility.await().untilAsserted(()
+                -> assertEquals(admin.topics().getReplicationClusters(topic, false), Collections.emptyList()));
+
         assertThrows(PulsarAdminException.PreconditionFailedException.class, () -> admin.topics()
                 .setReplicationClusters(topic, null));