From c729462f1c51ebeeb2ba050930be61e688b7e4ad Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=C3=93mar=20K=2E=20Yasin?= Date: Mon, 18 Aug 2025 12:45:56 -0700 Subject: [PATCH] [fix][broker] Allow empty replication clusters for namespace policies MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This change removes validation that prevented setting empty replication clusters for namespace policies. Previously, the API would reject empty cluster lists, but this is a valid configuration for disabling inter-cluster replication while maintaining topic accessibility within individual clusters. Changes: - Remove CollectionUtils.isEmpty(clusterIds) validation in NamespacesBase.java - Remove empty cluster validation in PersistentTopicsBase.java for topic-level policies - Add comprehensive test coverage for empty replication clusters in V1 and V2 namespace APIs - Verify bidirectional functionality (empty ↔ non-empty cluster configurations) - Ensure no auto-population occurs when explicitly setting empty clusters --- .../broker/admin/impl/NamespacesBase.java | 3 -- .../admin/impl/PersistentTopicsBase.java | 4 -- .../pulsar/broker/admin/NamespacesTest.java | 18 +++++++ .../pulsar/broker/admin/NamespacesV2Test.java | 52 ++++++++++++++++++- .../broker/admin/TopicPoliciesTest.java | 7 ++- 5 files changed, 74 insertions(+), 10 deletions(-) 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));