[fix][broker]Namespaces can be created with may empty replication_clusters policy (#25551)

Author: poorbarcode

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

@@ -491,7 +491,7 @@ private CompletableFuture<Policies> precheckWhenDeleteNamespace(NamespaceName ns
                                 // There are still more than one clusters configured for the global namespace
                                 throw new RestException(Status.PRECONDITION_FAILED,
                                     "Cannot delete the global namespace " + nsName + ". There are still more than "
-                                    + "one replication clusters configured.");
+                                    + "one replication clusters configured or replication clusters is empty.");
                             }
                             if (!cluster.equals(config().getClusterName())) {
                                 // the only replication cluster is other cluster, redirect
@@ -2392,7 +2392,7 @@ private CompletableFuture<Void> validatePoliciesAsync(NamespaceName ns, Policies
             log.info(msg);
             return FutureUtil.failedFuture(new RestException(Status.BAD_REQUEST, msg));
         }
-        pulsar().getBrokerService().setCurrentClusterAllowedIfNoClusterIsAllowed(ns, policies);
+        pulsar().getBrokerService().setCurrentClusterAllowedWhenCreating(ns, policies);
 
         // Validate cluster names and permissions
         return Stream.concat(policies.replication_clusters.stream(), policies.allowed_clusters.stream())

pulsar-broker/src/main/java/org/apache/pulsar/broker/service/BrokerService.java

@@ -4153,14 +4153,14 @@ public boolean isCurrentClusterAllowed(NamespaceName nsName, Policies nsPolicies
         return nsPolicies.replication_clusters.contains(pulsar.getConfig().getClusterName());
     }
 
-    public void setCurrentClusterAllowedIfNoClusterIsAllowed(NamespaceName nsName, Policies nsPolicies) {
-        if (nsPolicies.replication_clusters.contains(pulsar.getConfig().getClusterName())
-                || nsPolicies.allowed_clusters.contains(pulsar.getConfig().getClusterName())) {
-            return;
-        }
+    public void setCurrentClusterAllowedWhenCreating(NamespaceName nsName, Policies nsPolicies) {
         if (nsPolicies.replication_clusters.isEmpty()) {
             nsPolicies.replication_clusters.add(pulsar.getConfig().getClusterName());
-        } else {
+        }
+        if (nsPolicies.allowed_clusters.isEmpty()) {
+            return;
+        }
+        if (!nsPolicies.allowed_clusters.contains(pulsar.getConfig().getClusterName())) {
             nsPolicies.allowed_clusters.add(pulsar.getConfig().getClusterName());
         }
     }

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

@@ -41,7 +41,11 @@
 import com.google.common.collect.Lists;
 import com.google.common.collect.Sets;
 import java.lang.reflect.Field;
+import java.net.URI;
 import java.net.URL;
+import java.net.http.HttpClient;
+import java.net.http.HttpRequest;
+import java.net.http.HttpResponse;
 import java.nio.charset.StandardCharsets;
 import java.time.Clock;
 import java.util.ArrayList;
@@ -137,6 +141,7 @@
 import org.apache.pulsar.common.policies.data.PartitionedTopicStats;
 import org.apache.pulsar.common.policies.data.PersistencePolicies;
 import org.apache.pulsar.common.policies.data.PersistentTopicInternalStats;
+import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
 import org.apache.pulsar.common.policies.data.SubscriptionStats;
 import org.apache.pulsar.common.policies.data.TenantInfoImpl;
@@ -1739,6 +1744,36 @@ public void testCreateNamespaceWithNoClusters() throws PulsarAdminException {
                 Collections.singletonList(localCluster));
     }
 
+    @Test
+    public void testCreateNamespaceWithEmptyReplicationClustersByHttp() throws Exception {
+        String localCluster = pulsar.getConfiguration().getClusterName();
+        String namespacePart = newUniqueName("ns");
+        String namespace = defaultTenant + "/" + namespacePart;
+
+        // Create namespace with "allowed_cluster", and the param "replication_clusters" is empty.
+        HttpClient httpClient = HttpClient.newHttpClient();
+        URI adminV2Uri = URI.create(brokerUrl.toString()).resolve("/admin/v2/");
+        String namespaceRequestBody = "{\"allowed_clusters\": [\"" + localCluster + "\"]}";
+        HttpRequest createNamespaceRequest =
+                HttpRequest.newBuilder(adminV2Uri.resolve("namespaces/" + namespace))
+                        .header("Content-Type", "application/json")
+                        .PUT(HttpRequest.BodyPublishers.ofString(namespaceRequestBody))
+                        .build();
+        HttpResponse<String> createNamespaceResponse = httpClient.send(createNamespaceRequest,
+                HttpResponse.BodyHandlers.ofString());
+        assertEquals(createNamespaceResponse.statusCode(), Status.NO_CONTENT.getStatusCode(),
+                "Failed to create namespace by HTTP: " + createNamespaceResponse.body());
+
+        // Verify: replication_clusters is not empty.
+        Awaitility.await().untilAsserted(() -> {
+            Policies policies = admin.namespaces().getPolicies(namespace);
+            assertEquals(policies.replication_clusters.size(), 1);
+            assertEquals(policies.allowed_clusters.size(), 1);
+            assertTrue(policies.replication_clusters.contains(localCluster));
+            assertTrue(policies.allowed_clusters.contains(localCluster));
+        });
+    }
+
     @Test(timeOut = 30000)
     public void testConsumerStatsLastTimestamp() throws PulsarClientException, PulsarAdminException,
             InterruptedException {