If no clusters are specified, the namespace should by default use cur…

…rent cluster (#3571) * If no clusters are specified, the namespace should by default use current cluster * Don't apply to v1 local namespaces * It's better to use the isV2() check instead of isGlobal()
apache · Feb 14, 2019 · 48e4a0b · 48e4a0b
1 parent 14a5962
commit 48e4a0b
Show file tree

Hide file tree

Showing 2 changed files with 20 additions and 2 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
@@ -30,6 +30,7 @@
 
 import java.net.URI;
 import java.net.URL;
+import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 import java.util.Map.Entry;
@@ -74,9 +75,9 @@
 import org.apache.pulsar.common.policies.data.PersistencePolicies;
 import org.apache.pulsar.common.policies.data.Policies;
 import org.apache.pulsar.common.policies.data.RetentionPolicies;
+import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
 import org.apache.pulsar.common.policies.data.SubscribeRate;
 import org.apache.pulsar.common.policies.data.SubscriptionAuthMode;
-import org.apache.pulsar.common.policies.data.SchemaAutoUpdateCompatibilityStrategy;
 import org.apache.pulsar.common.util.FutureUtil;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.data.Stat;
@@ -404,7 +405,7 @@ protected void internalRevokePermissionsOnSubscription(String subscriptionName,
             throw new RestException(Status.NOT_IMPLEMENTED, "Authorization is not enabled");
         }
     }
-    
+
     protected Set<String> internalGetNamespaceReplicationClusters() {
         if (!namespaceName.isGlobal()) {
             throw new RestException(Status.PRECONDITION_FAILED,
@@ -1489,6 +1490,11 @@ public static BundlesData getBundles(int numBundles) {
     }
 
     private void validatePolicies(NamespaceName ns, Policies policies) {
+        if (ns.isV2() && policies.replication_clusters.isEmpty()) {
+            // Default to local cluster
+            policies.replication_clusters = Collections.singleton(config().getClusterName());
+        }
+
         // Validate cluster names and permissions
         policies.replication_clusters.forEach(cluster -> validateClusterForTenant(ns.getTenant(), cluster));
 

diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/admin/AdminApiTest2.java
@@ -31,6 +31,7 @@
 
 import java.net.URL;
 import java.util.ArrayList;
+import java.util.Collections;
 import java.util.HashMap;
 import java.util.HashSet;
 import java.util.List;
@@ -903,4 +904,15 @@ public void testClusterIsReadyBeforeCreateTopic() throws PulsarAdminException {
         } catch (PulsarAdminException.PreconditionFailedException e) {
         }
     }
+
+    @Test
+    public void testCreateNamespaceWithNoClusters() throws PulsarAdminException {
+        String localCluster = pulsar.getConfiguration().getClusterName();
+        String namespace = "prop-xyz/test-ns-with-no-clusters";
+        admin.namespaces().createNamespace(namespace);
+
+        // Global cluster, if there, should be omitted from the results
+        assertEquals(admin.namespaces().getNamespaceReplicationClusters(namespace),
+                Collections.singletonList(localCluster));
+    }
 }