[GitHub] [pulsar] Technoboy- commented on a change in pull request #11355: [Issue 11339] Pulsar Admin List Subscription lists only subscriptions created for Partition-0 when partition specific subscriptions are created

[GitBox <gi...@apache.org>]

Technoboy- commented on a change in pull request #11355:
URL: https://github.com/apache/pulsar/pull/11355#discussion_r671947387



##########
File path: pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/PersistentTopicsBase.java
##########
@@ -1030,34 +1030,46 @@ protected void internalGetSubscriptions(AsyncResponse asyncResponse, boolean aut
                     false).thenAccept(partitionMetadata -> {
                 if (partitionMetadata.partitions > 0) {
                     try {
-                        // get the subscriptions only from the 1st partition
-                        // since all the other partitions will have the same
-                        // subscriptions
-                        pulsar().getAdminClient().topics().getSubscriptionsAsync(topicName.getPartition(0).toString())
-                                .whenComplete((r, ex) -> {
-                                    if (ex != null) {
-                                        log.warn("[{}] Failed to get list of subscriptions for {}: {}", clientAppId(),
-                                                topicName, ex.getMessage());
-
-                                        if (ex instanceof PulsarAdminException) {
-                                            PulsarAdminException pae = (PulsarAdminException) ex;
-                                            if (pae.getStatusCode() == Status.NOT_FOUND.getStatusCode()) {
-                                                asyncResponse.resume(new RestException(Status.NOT_FOUND,
-                                                        "Internal topics have not been generated yet"));
-                                                return;
-                                            } else {
-                                                asyncResponse.resume(new RestException(pae));
-                                                return;
-                                            }
-                                        } else {
-                                            asyncResponse.resume(new RestException(ex));
-                                            return;
-                                        }
+                        final Set<String> subscriptions = Sets.newConcurrentHashSet();
+                        final List<CompletableFuture<Object>> subscriptionFutures = Lists.newArrayList();
+                        String path = String.format("/managed-ledgers/%s/%s", namespaceName.toString(), domain());
+                        List<String> children = getLocalPolicies().getChildren(path);
+                        List<String> activeTopics = Lists.newArrayList();
+                        for (String topic : children) {
+                            if (topic.contains(topicName.getLocalName())) {
+                                activeTopics.add(topic);
+                            }
+                        }
+                        if (log.isDebugEnabled()) {
+                            log.debug("activeTopics : {}", activeTopics);
+                        }
+                        for (String topic : activeTopics) {
+                            CompletableFuture<List<String>> subscriptionsAsync = pulsar().getAdminClient().topics()
+                                    .getSubscriptionsAsync(TopicName.get(domain(), namespaceName, topic).toString());
+                            subscriptionFutures.add(subscriptionsAsync.thenApply(r -> subscriptions.addAll(r)));

Review comment:
       Thanks, replace with stream and lambda.




-- 
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.

To unsubscribe, e-mail: commits-unsubscribe@pulsar.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org