[Issue 7846]Add support for regex patterns in tenant and namespace

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

@@ -30,6 +30,7 @@
 import java.lang.reflect.Field;
 import java.net.URI;
 import java.net.URL;
+import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
 import java.util.Map;
@@ -41,6 +42,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 import java.util.function.Function;
+import java.util.regex.Pattern;
 import java.util.stream.Collectors;
 import javax.ws.rs.WebApplicationException;
 import javax.ws.rs.container.AsyncResponse;
@@ -105,6 +107,34 @@ public abstract class NamespacesBase extends AdminResource {
 
     private static final long MAX_BUNDLES = ((long) 1) << 32;
 
+    protected List<String> internalGetNamespaces(String regex) {
+        ArrayList<String> result = new ArrayList<>();
+        Pattern p = Pattern.compile(regex);
+        try {
+            List<String> tenants = globalZk().getChildren(path(POLICIES), false);
+            for (String tenant : tenants) {
+                validateTenantOperation(tenant, TenantOperation.LIST_NAMESPACES);
+                result.addAll(
+                    getListOfNamespaces(tenant).stream()
+                        .filter(namespace -> {
+                            try {
+                                validateNamespaceOperation(NamespaceName.get(namespace), NamespaceOperation.GET_TOPICS);
+                                return true;
+                            } catch (Exception e){
+                                return false;
+                            }
+                        })
+                        .filter(namespace -> p.matcher(namespace).matches())
+                        .collect(Collectors.toList())
+                );
+            }
+        } catch (Exception e) {
+            log.error("[{}] Failed to get tenants list", clientAppId(), e);
+            throw new RestException(e);
+        }
+        return result;
+    }
+
     protected List<String> internalGetTenantNamespaces(String tenant) {
         checkNotNull(tenant, "Tenant should not be null");
         try {

pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/v2/Namespaces.java

@@ -80,6 +80,15 @@
 @Api(value = "/namespaces", description = "Namespaces admin apis", tags = "namespaces")
 public class Namespaces extends NamespacesBase {
 
+    @GET
+    @Path("")
+    @ApiOperation(value = "Get all namespaces filtered by query params")
+    @ApiResponses(value = { @ApiResponse(code = 403, message = "Don't have admin permission"),
+            @ApiResponse(code = 404, message = "No matching namespaces") })
+    public List<String> getNamespaces(@QueryParam("regex") @DefaultValue(".*") String regex) {
+        return internalGetNamespaces(regex);
+    }
+
     @GET
     @Path("/{tenant}")
     @ApiOperation(value = "Get the list of all the namespaces for a certain tenant.", response = String.class, responseContainer = "Set")

pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java

@@ -70,6 +70,7 @@
 import org.apache.zookeeper.AsyncCallback.StatCallback;
 import org.apache.zookeeper.KeeperException;
 import org.apache.zookeeper.KeeperException.Code;
+import org.apache.zookeeper.ZooKeeper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -1031,6 +1032,57 @@ public CompletableFuture<List<String>> getListOfTopics(NamespaceName namespaceNa
         }
     }
 
+    public List<String> getNamespacesByRegex(String regex) throws Exception{
+        ArrayList<String> result = new ArrayList<>();
+        Pattern p = Pattern.compile(regex);
+
+        ZooKeeper globalZk = pulsar.getGlobalZkCache().getZooKeeper();
+        try {
+            List<String> tenants = globalZk.getChildren(AdminResource.path(POLICIES), false);
+            for (String tenant : tenants) {
+                result.addAll(
+                        getListOfNamespaces(tenant).stream()
+                                .filter(namespace -> p.matcher(namespace).matches())
+                                .collect(Collectors.toList())
+                );
+            }
+        } catch (Exception e) {
+            LOG.error("Failed to get tenants list", e);
+            throw new Exception(e);
+        }
+        return result;
+    }
+
+    private List<String> getListOfNamespaces(String property) throws Exception {
+        List<String> namespaces = Lists.newArrayList();
+
+        ZooKeeper globalZk = pulsar.getGlobalZkCache().getZooKeeper();
+        // this will return a cluster in v1 and a namespace in v2
+        for (String clusterOrNamespace : globalZk.getChildren(AdminResource.path(POLICIES, property), false)) {
+            // Then get the list of namespaces
+            try {
+                final List<String> children = globalZk.getChildren(AdminResource.path(POLICIES, property, clusterOrNamespace), false);
+                if (children == null || children.isEmpty()) {
+                    String namespace = NamespaceName.get(property, 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
+                    if (globalZk.getData(AdminResource.path(POLICIES, namespace), false, null).length != 0) {
+                        namespaces.add(namespace);
+                    }
+                } else {
+                    children.forEach(ns -> {
+                        namespaces.add(NamespaceName.get(property, clusterOrNamespace, ns).toString());
+                    });
+                }
+            } catch (KeeperException.NoNodeException e) {
+                // A cluster was deleted between the 2 getChildren() calls, ignoring
+            }
+        }
+
+        namespaces.sort(null);
+        return namespaces;
+    }
+
     public CompletableFuture<List<String>> getAllPartitions(NamespaceName namespaceName) {
         return getPartitions(namespaceName, TopicDomain.persistent)
                 .thenCombine(getPartitions(namespaceName, TopicDomain.non_persistent),

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

@@ -92,6 +92,7 @@
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetOrCreateSchema;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetSchema;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespace;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetNamespacesByRegex;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandLookupTopic;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandNewTxn;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandPartitionedTopicMetadata;
@@ -1575,6 +1576,30 @@ protected void handleGetTopicsOfNamespace(CommandGetTopicsOfNamespace commandGet
                 });
     }
 
+    @Override
+    protected void handleGetNamespaceByRegex(CommandGetNamespacesByRegex commandGetNamespaceByRegex) {
+        final long requestId = commandGetNamespaceByRegex.getRequestId();
+        final String regex = commandGetNamespaceByRegex.getRegex();
+
+        try {
+            List<String> namespaces = getBrokerService().pulsar().getNamespaceService().getNamespacesByRegex(regex);
+            if (log.isDebugEnabled()) {
+                log.debug("[{}] Received CommandGetTopicsOfNamespace for regex [//{}] by {}, size:{}",
+                        remoteAddress, regex, requestId, namespaces.size());
+            }
+
+            ctx.writeAndFlush(Commands.newGetNamespacesByRegexResponse(namespaces, requestId));
+
+        } catch (Exception ex) {
+            log.warn("[{}] Error GetNamespaceByRegex for regex [//{}] by {}",
+                    remoteAddress, regex, requestId);
+            ctx.writeAndFlush(
+                    Commands.newError(requestId,
+                            BrokerServiceException.getClientErrorCode(new ServerMetadataException(ex)),
+                            ex.getMessage()));
+        }
+    }
+
     @Override
     protected void handleGetSchema(CommandGetSchema commandGetSchema) {
         if (log.isDebugEnabled()) {

pulsar-broker/src/test/java/org/apache/pulsar/client/api/ProducerConsumerBase.java

@@ -43,6 +43,8 @@ public void producerBaseSetup() throws Exception {
                 new TenantInfo(Sets.newHashSet("appid1", "appid2"), Sets.newHashSet("test")));
         admin.namespaces().createNamespace("my-property/my-ns");
         admin.namespaces().setNamespaceReplicationClusters("my-property/my-ns", Sets.newHashSet("test"));
+        admin.namespaces().createNamespace("my-property/my-ns1");
+        admin.namespaces().setNamespaceReplicationClusters("my-property/my-ns1", Sets.newHashSet("test"));
 
         // so that clients can test short names
         admin.tenants().createTenant("public",

pulsar-broker/src/test/java/org/apache/pulsar/client/impl/PatternTopicsConsumerImplTest.java

@@ -561,8 +561,8 @@ public void testAutoSubscribePatternConsumer() throws Exception {
         String subscriptionName = "my-ex-subscription-" + key;
         String topicName1 = "persistent://my-property/my-ns/pattern-topic-1-" + key;
         String topicName2 = "persistent://my-property/my-ns/pattern-topic-2-" + key;
-        String topicName3 = "persistent://my-property/my-ns/pattern-topic-3-" + key;
-        Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*");
+        String topicName3 = "persistent://my-property/my-ns1/pattern-topic-3-" + key;
+        Pattern pattern = Pattern.compile("persistent://my-property/my-ns\\d*/pattern-topic.*");
 
         // 1. create partition
         TenantInfo tenantInfo = createDefaultTenantInfo();
@@ -665,13 +665,13 @@ public void testAutoSubscribePatternConsumer() throws Exception {
     }
 
     @Test(timeOut = testTimeout)
-    public void testAutoUnbubscribePatternConsumer() throws Exception {
+    public void testAutoUnsubscribePatternConsumer() throws Exception {
         String key = "AutoUnsubscribePatternConsumer";
         String subscriptionName = "my-ex-subscription-" + key;
         String topicName1 = "persistent://my-property/my-ns/pattern-topic-1-" + key;
         String topicName2 = "persistent://my-property/my-ns/pattern-topic-2-" + key;
-        String topicName3 = "persistent://my-property/my-ns/pattern-topic-3-" + key;
-        Pattern pattern = Pattern.compile("persistent://my-property/my-ns/pattern-topic.*");
+        String topicName3 = "persistent://my-property/my-ns1/pattern-topic-3-" + key;
+        Pattern pattern = Pattern.compile("persistent://my-property/my-ns\\d*/pattern-topic.*");
 
         // 1. create partition
         TenantInfo tenantInfo = createDefaultTenantInfo();
@@ -737,6 +737,8 @@ public void testAutoUnbubscribePatternConsumer() throws Exception {
         NamespaceService nss = pulsar.getNamespaceService();
         doReturn(CompletableFuture.completedFuture(topicNames)).when(nss)
                 .getListOfPersistentTopics(NamespaceName.get("my-property/my-ns"));
+        doReturn(CompletableFuture.completedFuture(topicNames)).when(nss)
+                .getListOfPersistentTopics(NamespaceName.get("my-property/my-ns1"));
 
         // 7. call recheckTopics to unsubscribe topic 1,3 , verify topics number: 2=6-1-3
         log.debug("recheck topics change");

pulsar-client/src/main/java/org/apache/pulsar/client/impl/BinaryProtoLookupService.java

@@ -312,6 +312,65 @@ private void getTopicsUnderNamespace(InetSocketAddress socketAddress,
         });
     }
 
+    @Override
+    public CompletableFuture<List<NamespaceName>> getNamespacesByRegex(String regex) {
+        CompletableFuture<List<NamespaceName>> topicsFuture = new CompletableFuture<List<NamespaceName>>();
+
+        AtomicLong opTimeoutMs = new AtomicLong(client.getConfiguration().getOperationTimeoutMs());
+        Backoff backoff = new BackoffBuilder()
+                .setInitialTime(100, TimeUnit.MILLISECONDS)
+                .setMandatoryStop(opTimeoutMs.get() * 2, TimeUnit.MILLISECONDS)
+                .setMax(1, TimeUnit.MINUTES)
+                .create();
+        getNamespacesByRegex(serviceNameResolver.resolveHost(), regex, backoff, opTimeoutMs, topicsFuture);
+        return topicsFuture;
+    }
+
+    private void getNamespacesByRegex(InetSocketAddress socketAddress,
+                                          String regex,
+                                          Backoff backoff,
+                                          AtomicLong remainingTime,
+                                          CompletableFuture<List<NamespaceName>> namespacesFuture) {
+        client.getCnxPool().getConnection(socketAddress).thenAccept(clientCnx -> {
+            long requestId = client.newRequestId();
+            ByteBuf request = Commands.newGetNamespacesByRegexRequest(regex, requestId);
+
+            clientCnx.newGetNamespaceByRegex(request, requestId).whenComplete((r, n) -> {
+                if (n != null) {
+                    namespacesFuture.completeExceptionally(n);
+                } else {
+                    if (log.isDebugEnabled()) {
+                        log.debug("[regex: {}] Success get topics list in request: {}", regex, requestId);
+                    }
+
+                    List<NamespaceName> result = Lists.newArrayList();
+                    r.forEach(topic -> {
+                        result.add(NamespaceName.get(topic));
+                    });
+
+                    namespacesFuture.complete(result);
+                }
+                client.getCnxPool().releaseConnection(clientCnx);
+            });
+        }).exceptionally((e) -> {
+            long nextDelay = Math.min(backoff.next(), remainingTime.get());
+            if (nextDelay <= 0) {
+                namespacesFuture.completeExceptionally(
+                        new PulsarClientException.TimeoutException(
+                                format("Could not get namespaces for regex %s within configured timeout",
+                                        regex)));
+                return null;
+            }
+
+            ((ScheduledExecutorService) executor).schedule(() -> {
+                log.warn("[regex: {}] Could not get connection while getNamespacesByRegex -- Will try again in {} ms",
+                        regex, nextDelay);
+                remainingTime.addAndGet(-nextDelay);
+                getNamespacesByRegex(socketAddress, regex, backoff, remainingTime, namespacesFuture);
+            }, nextDelay, TimeUnit.MILLISECONDS);
+            return null;
+        });
+    }
 
     @Override
     public void close() throws Exception {

pulsar-client/src/main/java/org/apache/pulsar/client/impl/ClientCnx.java

@@ -71,6 +71,7 @@
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandConnected;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandError;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetLastMessageIdResponse;
+import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetNamespacesByRegexResponse;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetSchemaResponse;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetOrCreateSchemaResponse;
 import org.apache.pulsar.common.api.proto.PulsarApi.CommandGetTopicsOfNamespaceResponse;
@@ -713,6 +714,42 @@ protected void handleGetTopicsOfNamespaceSuccess(CommandGetTopicsOfNamespaceResp
         }
     }
 
+    public CompletableFuture<List<String>> newGetNamespaceByRegex(ByteBuf request, long requestId) {
+        CompletableFuture<List<String>> future = new CompletableFuture<>();
+
+        pendingGetTopicsRequests.put(requestId, future);
+        ctx.writeAndFlush(request).addListener(writeFuture -> {
+            if (!writeFuture.isSuccess()) {
+                log.warn("{} Failed to send request {} to broker: {}", ctx.channel(), requestId,
+                        writeFuture.cause().getMessage());
+                pendingGetTopicsRequests.remove(requestId);
+                future.completeExceptionally(writeFuture.cause());
+            }
+        });
+
+        return future;
+    }
+
+    @Override
+    protected void handleGetNamespaceByRegexSuccess(CommandGetNamespacesByRegexResponse success) {
+        checkArgument(state == State.Ready);
+
+        long requestId = success.getRequestId();
+        List<String> namespaces = success.getNamespacesList();
+
+        if (log.isDebugEnabled()) {
+            log.debug("{} Received get namespaces by regex success response from server: {} - namespaces.size: {}",
+                    ctx.channel(), success.getRequestId(), namespaces.size());
+        }
+
+        CompletableFuture<List<String>> requestFuture = pendingGetTopicsRequests.remove(requestId);
+        if (requestFuture != null) {
+            requestFuture.complete(namespaces);
+        } else {
+            log.warn("{} Received unknown request id from server: {}", ctx.channel(), success.getRequestId());
+        }
+    }
+
     @Override
     protected void handleGetSchemaResponse(CommandGetSchemaResponse commandGetSchemaResponse) {
         checkArgument(state == State.Ready);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/HttpLookupService.java

@@ -136,6 +136,26 @@ public CompletableFuture<List<String>> getTopicsUnderNamespace(NamespaceName nam
         return future;
     }
 
+    @Override
+    public CompletableFuture<List<NamespaceName>> getNamespacesByRegex(String regex) {
+        CompletableFuture<List<NamespaceName>> future = new CompletableFuture<>();
+
+        httpClient
+                .get(String.format("admin/v2/namespaces?regex=%s", regex), String[].class)
+                .thenAccept(namespaces -> {
+                    List<NamespaceName> result = Lists.newArrayList();
+                    Arrays.asList(namespaces).forEach(namespace -> {
+                        result.add(NamespaceName.get(namespace));
+                    });
+                    future.complete(result);})
+                .exceptionally(ex -> {
+                    log.warn("Failed to getNamespacesByRegex regex {} {} .", regex, ex.getMessage());
+                    future.completeExceptionally(ex);
+                    return null;
+                });
+        return future;
+    }
+
     @Override
     public CompletableFuture<Optional<SchemaInfo>> getSchema(TopicName topicName) {
         return getSchema(topicName, null);

pulsar-client/src/main/java/org/apache/pulsar/client/impl/LookupService.java

@@ -99,4 +99,11 @@ public interface LookupService extends AutoCloseable {
 	 */
 	CompletableFuture<List<String>> getTopicsUnderNamespace(NamespaceName namespace, Mode mode);
 
+	/**
+	 * Returns the names of all namespaces for a given tenant.
+	 *
+	 * @return
+	 */
+	CompletableFuture<List<NamespaceName>> getNamespacesByRegex(String regex);
+
 }