[fix] PIP-457: Remove support for V1 topic names and V1 Admin API (implementation)

bouncy-castle/bcfips-include-test/src/test/java/org/apache/pulsar/client/TlsProducerConsumerTest.java

@@ -48,10 +48,10 @@ public void testTlsLargeSizeMessage() throws Exception {
         internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
         internalSetUpForNamespace();
 
-        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
+        Consumer<byte[]> consumer = pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
                 .subscriptionName("my-subscriber-name").subscribe();
 
-        Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/use/my-ns/my-topic1")
+        Producer<byte[]> producer = pulsarClient.newProducer().topic("persistent://my-property/my-ns/my-topic1")
                 .create();
         for (int i = 0; i < 10; i++) {
             byte[] message = new byte[messageSize];
@@ -83,7 +83,7 @@ public void testTlsClientAuthOverBinaryProtocol() throws Exception {
         // Test 1 - Using TLS on binary protocol without sending certs - expect failure
         internalSetUpForClient(false, pulsar.getBrokerServiceUrlTls());
         try {
-            pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
+            pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
             Assert.fail("Server should have failed the TLS handshake since client didn't .");
         } catch (Exception ex) {
@@ -94,7 +94,7 @@ public void testTlsClientAuthOverBinaryProtocol() throws Exception {
         // Test 2 - Using TLS on binary protocol - sending certs
         internalSetUpForClient(true, pulsar.getBrokerServiceUrlTls());
         try {
-            pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
+            pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
             log.info("second test success: with certs set, consumer sub success");
         } catch (Exception ex) {
@@ -113,7 +113,7 @@ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
         // Test 1 - Using TLS on https without sending certs - expect failure
         internalSetUpForClient(false, pulsar.getWebServiceAddressTls());
         try {
-            pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
+            pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
             Assert.fail("Server should have failed the TLS handshake since client didn't .");
         } catch (Exception ex) {
@@ -124,7 +124,7 @@ public void testTlsClientAuthOverHTTPProtocol() throws Exception {
         // Test 2 - Using TLS on https - sending certs
         internalSetUpForClient(true, pulsar.getWebServiceAddressTls());
         try {
-            pulsarClient.newConsumer().topic("persistent://my-property/use/my-ns/my-topic1")
+            pulsarClient.newConsumer().topic("persistent://my-property/my-ns/my-topic1")
                     .subscriptionName("my-subscriber-name").subscriptionType(SubscriptionType.Exclusive).subscribe();
             log.info("second test success: with certs set, consumer sub success");
         } catch (Exception ex) {

conf/broker.conf

@@ -198,10 +198,6 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
-# If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',
-# the topic cannot be automatically created.
-allowAutoTopicCreationWithLegacyNamingScheme=true
-
 # If 'strictSubscriptionNameVerification' is true, the new subscription name can only contain (a-zA-Z_0-9) and these
 # special chars -=:.
 strictlyVerifySubscriptionName=false

conf/standalone.conf

@@ -1311,10 +1311,6 @@ allowAutoTopicCreation=true
 # The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)
 allowAutoTopicCreationType=non-partitioned
 
-# If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',
-# the topic cannot be automatically created.
-allowAutoTopicCreationWithLegacyNamingScheme=true
-
 # If 'strictSubscriptionNameVerification' is true, the new subscription name can only contain (a-zA-Z_0-9) and these
 # special chars -=:.
 strictlyVerifySubscriptionName=false

managed-ledger/src/test/java/org/apache/bookkeeper/mledger/impl/ManagedLedgerBkTest.java

@@ -542,7 +542,7 @@ public void testOfflineTopicBacklog() throws Exception {
         ManagedLedgerConfig config = new ManagedLedgerConfig();
         config.setEnsembleSize(1).setWriteQuorumSize(1).setAckQuorumSize(1).setMetadataEnsembleSize(1)
                 .setMetadataAckQuorumSize(1);
-        ManagedLedger ledger = factory.open("property/cluster/namespace/my-ledger", config);
+        ManagedLedger ledger = factory.open("property/namespace/my-ledger", config);
         ManagedCursor cursor = ledger.openCursor("c1");
 
         int num = 1;
@@ -560,7 +560,7 @@ public void testOfflineTopicBacklog() throws Exception {
         ManagedLedgerOfflineBacklog offlineTopicBacklog = new ManagedLedgerOfflineBacklog(
                 DigestType.CRC32, "".getBytes(StandardCharsets.UTF_8), "", false);
         PersistentOfflineTopicStats offlineTopicStats = offlineTopicBacklog.getEstimatedUnloadedTopicBacklog(
-                (ManagedLedgerFactoryImpl) factory, "property/cluster/namespace/my-ledger");
+                (ManagedLedgerFactoryImpl) factory, "property/namespace/my-ledger");
         assertNotNull(offlineTopicStats);
     }
 

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java

@@ -2360,11 +2360,6 @@ The max allowed delay for delayed delivery (in milliseconds). If the broker rece
             doc = "The type of topic that is allowed to be automatically created.(partitioned/non-partitioned)"
     )
     private TopicType allowAutoTopicCreationType = TopicType.NON_PARTITIONED;
-    @FieldContext(category = CATEGORY_SERVER, dynamic = true,
-            doc = "If 'allowAutoTopicCreation' is true and the name of the topic contains 'cluster',"
-                    + "the topic cannot be automatically created."
-    )
-    private boolean allowAutoTopicCreationWithLegacyNamingScheme = true;
     @FieldContext(category = CATEGORY_SERVER, dynamic = true,
             doc = "If 'strictSubscriptionNameVerification' is true, the new subscription name can only contain"
                 + " (a-zA-Z_0-9) and these special chars -=:."

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/authorization/PulsarAuthorizationProvider.java

@@ -462,19 +462,7 @@ private CompletableFuture<Void> updateSubscriptionPermissionAsync(NamespaceName
     }
 
     private CompletableFuture<Boolean> checkAuthorization(TopicName topicName, String role, AuthAction action) {
-        return checkPermission(topicName, role, action).thenCompose(permission ->
-                permission ? checkCluster(topicName) : CompletableFuture.completedFuture(false));
-    }
-
-    private CompletableFuture<Boolean> checkCluster(TopicName topicName) {
-        if (topicName.isGlobal() || conf.getClusterName().equals(topicName.getCluster())) {
-            return CompletableFuture.completedFuture(true);
-        }
-        if (log.isDebugEnabled()) {
-            log.debug("Topic [{}] does not belong to local cluster [{}]", topicName.toString(), conf.getClusterName());
-        }
-        return pulsarResources.getClusterResources().listAsync()
-                .thenApply(clusters -> clusters.contains(topicName.getCluster()));
+        return checkPermission(topicName, role, action);
     }
 
     public CompletableFuture<Boolean> checkPermission(TopicName topicName, String role, AuthAction action) {

pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TenantResources.java

@@ -19,7 +19,6 @@
 package org.apache.pulsar.broker.resources;
 
 import java.util.ArrayList;
-import java.util.Collections;
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
@@ -48,8 +47,8 @@ public CompletableFuture<List<String>> listTenantsAsync() {
 
     public CompletableFuture<Void> deleteTenantAsync(String tenantName) {
         return getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenantName))
-                .thenCompose(clusters -> FutureUtil.waitForAll(clusters.stream()
-                        .map(cluster -> getCache().delete(joinPath(BASE_POLICIES_PATH, tenantName, cluster)))
+                .thenCompose(namespaces -> FutureUtil.waitForAll(namespaces.stream()
+                        .map(ns -> getCache().delete(joinPath(BASE_POLICIES_PATH, tenantName, ns)))
                         .collect(Collectors.toList()))
                 ).thenCompose(__ -> deleteAsync(joinPath(BASE_POLICIES_PATH, tenantName)));
     }
@@ -85,124 +84,79 @@ public CompletableFuture<Boolean> tenantExistsAsync(String tenantName) {
     public List<String> getListOfNamespaces(String tenant) throws MetadataStoreException {
         List<String> namespaces = new ArrayList<>();
 
-        // this will return a cluster in v1 and a namespace in v2
-        for (String clusterOrNamespace : getChildren(joinPath(BASE_POLICIES_PATH, tenant))) {
-            // Then get the list of namespaces
-            final List<String> children = getChildren(joinPath(BASE_POLICIES_PATH, tenant, clusterOrNamespace));
-            if (children == null || children.isEmpty()) {
-                String namespace = NamespaceName.get(tenant, clusterOrNamespace).toString();
-                // if the length is 0 then this is probably a leftover cluster from namespace created
-                // with the v1 admin format (prop/cluster/ns) and then deleted, so no need to add it to the list
-                try {
-                    if (get(joinPath(BASE_POLICIES_PATH, namespace)).isPresent()) {
-                        namespaces.add(namespace);
-                    }
-                } catch (MetadataStoreException.ContentDeserializationException e) {
-                    // not a namespace node
+        for (String ns : getChildren(joinPath(BASE_POLICIES_PATH, tenant))) {
+            String namespace = NamespaceName.get(tenant, ns).toString();
+            try {
+                if (get(joinPath(BASE_POLICIES_PATH, namespace)).isPresent()) {
+                    namespaces.add(namespace);
                 }
-
-            } else {
-                children.forEach(ns -> {
-                    namespaces.add(NamespaceName.get(tenant, clusterOrNamespace, ns).toString());
-                });
+            } catch (MetadataStoreException.ContentDeserializationException e) {
+                // not a namespace node
             }
         }
 
         return namespaces;
     }
 
     public CompletableFuture<List<String>> getListOfNamespacesAsync(String tenant) {
-        // this will return a cluster in v1 and a namespace in v2
         return getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant))
-                .thenCompose(clusterOrNamespaces -> clusterOrNamespaces.stream().map(key ->
-                        getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant, key))
-                                .thenCompose(children -> {
-                                    if (children == null || children.isEmpty()) {
-                                        String namespace = NamespaceName.get(tenant, key).toString();
-                                        // if the length is 0 then this is probably a leftover cluster from namespace
-                                        // created with the v1 admin format (prop/cluster/ns) and then deleted, so no
-                                        // need to add it to the list
-                                        return getAsync(joinPath(BASE_POLICIES_PATH, namespace))
-                                           .thenApply(opt -> opt.isPresent() ? Collections.singletonList(namespace)
-                                                   : new ArrayList<String>())
-                                           .exceptionally(ex -> {
-                                                Throwable cause = FutureUtil.unwrapCompletionException(ex);
-                                                if (cause instanceof MetadataStoreException
-                                                        .ContentDeserializationException) {
-                                                    return new ArrayList<>();
-                                                }
-                                                throw FutureUtil.wrapToCompletionException(ex);
-                                            });
-                                    } else {
-                                        CompletableFuture<List<String>> ret = new CompletableFuture();
-                                        ret.complete(children.stream().map(ns -> NamespaceName.get(tenant, key, ns)
-                                                .toString()).collect(Collectors.toList()));
-                                        return ret;
-                                    }
-                                })).reduce(CompletableFuture.completedFuture(new ArrayList<>()),
-                                        (accumulator, n) -> accumulator.thenCompose(namespaces -> n.thenApply(m -> {
-                                            namespaces.addAll(m);
-                                            return namespaces;
-                                    }))));
-    }
-
-    public CompletableFuture<List<String>> getActiveNamespaces(String tenant, String cluster) {
-        return getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant, cluster));
+                .thenCompose(nsList -> nsList.stream().map(key -> {
+                    String namespace = NamespaceName.get(tenant, key).toString();
+                    return getAsync(joinPath(BASE_POLICIES_PATH, namespace))
+                        .thenApply(opt -> opt.isPresent()
+                                ? List.of(namespace)
+                                : new ArrayList<String>())
+                        .exceptionally(ex -> {
+                            Throwable cause = FutureUtil.unwrapCompletionException(ex);
+                            if (cause instanceof MetadataStoreException
+                                    .ContentDeserializationException) {
+                                return new ArrayList<>();
+                            }
+                            throw FutureUtil.wrapToCompletionException(ex);
+                        });
+                }).reduce(CompletableFuture.completedFuture(new ArrayList<>()),
+                        (accumulator, n) -> accumulator.thenCompose(namespaces -> n.thenApply(m -> {
+                            namespaces.addAll(m);
+                            return namespaces;
+                        }))));
     }
 
     public CompletableFuture<Void> hasActiveNamespace(String tenant) {
         CompletableFuture<Void> activeNamespaceFuture = new CompletableFuture<>();
-        getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant)).thenAccept(clusterOrNamespaceList -> {
-            if (clusterOrNamespaceList == null || clusterOrNamespaceList.isEmpty()) {
+        getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant)).thenAccept(nsList -> {
+            if (nsList == null || nsList.isEmpty()) {
                 activeNamespaceFuture.complete(null);
                 return;
             }
             List<CompletableFuture<Void>> activeNamespaceListFuture = new ArrayList<>();
-            clusterOrNamespaceList.forEach(clusterOrNamespace -> {
-                // get list of active V1 namespace
+            nsList.forEach(ns -> {
                 CompletableFuture<Void> checkNs = new CompletableFuture<>();
                 activeNamespaceListFuture.add(checkNs);
-                getChildrenAsync(joinPath(BASE_POLICIES_PATH, tenant, clusterOrNamespace))
-                        .whenComplete((children, ex) -> {
-                            if (ex != null) {
-                                checkNs.completeExceptionally(ex);
-                                return;
-                            }
-                            if (children != null && !children.isEmpty()) {
-                                checkNs.completeExceptionally(
-                                        new IllegalStateException("The tenant still has active namespaces"));
-                                return;
-                            }
-                            String namespace = NamespaceName.get(tenant, clusterOrNamespace).toString();
-                            // if the length is 0 then this is probably a leftover cluster from namespace
-                            // created
-                            // with the v1 admin format (prop/cluster/ns) and then deleted, so no need to
-                            // add it to the list
-                            getAsync(joinPath(BASE_POLICIES_PATH, namespace)).thenApply(data -> {
-                                if (data.isPresent()) {
-                                    checkNs.completeExceptionally(new IllegalStateException(
-                                            "The tenant still has active namespaces"));
-                                } else {
-                                    checkNs.complete(null);
-                                }
-                                return null;
-                            }).exceptionally(ex2 -> {
-                                if (ex2.getCause() instanceof MetadataStoreException.ContentDeserializationException) {
-                                    // it's not a valid namespace-node
-                                    checkNs.complete(null);
-                                } else {
-                                    checkNs.completeExceptionally(ex2);
-                                }
-                                return null;
-                            });
-                        });
-                FutureUtil.waitForAll(activeNamespaceListFuture).thenAccept(r -> {
-                    activeNamespaceFuture.complete(null);
-                }).exceptionally(ex -> {
-                    activeNamespaceFuture.completeExceptionally(ex.getCause());
+                String namespace = NamespaceName.get(tenant, ns).toString();
+                getAsync(joinPath(BASE_POLICIES_PATH, namespace)).thenApply(data -> {
+                    if (data.isPresent()) {
+                        checkNs.completeExceptionally(new IllegalStateException(
+                                "The tenant still has active namespaces"));
+                    } else {
+                        checkNs.complete(null);
+                    }
+                    return null;
+                }).exceptionally(ex2 -> {
+                    if (ex2.getCause() instanceof MetadataStoreException.ContentDeserializationException) {
+                        // it's not a valid namespace-node
+                        checkNs.complete(null);
+                    } else {
+                        checkNs.completeExceptionally(ex2);
+                    }
                     return null;
                 });
             });
+            FutureUtil.waitForAll(activeNamespaceListFuture).thenAccept(r -> {
+                activeNamespaceFuture.complete(null);
+            }).exceptionally(ex -> {
+                activeNamespaceFuture.completeExceptionally(ex.getCause());
+                return null;
+            });
         }).exceptionally(ex -> {
             activeNamespaceFuture.completeExceptionally(ex.getCause());
             return null;