diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java index d393a71da727f..543d00d730bea 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/NamespacesBase.java @@ -755,9 +755,6 @@ protected CompletableFuture internalSetNamespaceReplicationClusters(List 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, 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 3b74e19d2884e..e02a262dd8776 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 @@ -3346,10 +3346,6 @@ protected CompletableFuture internalSetBacklogQuota(BacklogQuota.BacklogQu } protected CompletableFuture internalSetReplicationClusters(List clusterIds, boolean isGlobal) { - if (CollectionUtils.isEmpty(clusterIds)) { - return CompletableFuture.failedFuture(new RestException(Status.PRECONDITION_FAILED, - "ClusterIds should not be null or empty")); - } Set replicationClusters = Sets.newHashSet(clusterIds); return validatePoliciesReadOnlyAccessAsync() .thenAccept(__ -> { diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java index b46c1f92faa07..a275180dd9b74 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesTest.java +++ b/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 repClusters = (Set) 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 boundaries = List.of("0x00000000", "0x80000000", "0xffffffff"); diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java index c1e8dfa30994a..0ed2e030c48c2 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/NamespacesV2Test.java +++ b/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 repClusters = admin.namespaces().getNamespaceReplicationClusters(v2Namespace.toString()); + assertEquals(repClusters, Collections.emptyList()); + + // Clean up + admin.namespaces().deleteNamespace(v2Namespace.toString()); + } } diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java index 1d911c52c838c..a445c34453ea9 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/TopicPoliciesTest.java +++ b/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));