andrasbeni commented on code in PR #16901:
URL: https://github.com/apache/pulsar/pull/16901#discussion_r938769579
##########
pulsar-broker-common/src/main/java/org/apache/pulsar/broker/resources/TopicResources.java:
##########
@@ -41,22 +42,29 @@ public class TopicResources {
private static final String MANAGED_LEDGER_PATH = "/managed-ledgers";
private final MetadataStore store;
+ private final NamespaceResources namespaceResources;
private final Map<BiConsumer<String, NotificationType>, Pattern>
topicListeners;
- public TopicResources(MetadataStore store) {
+ public TopicResources(MetadataStore store, NamespaceResources
namespaceResources) {
this.store = store;
+ this.namespaceResources = namespaceResources;
topicListeners = new ConcurrentHashMap<>();
store.registerListener(this::handleNotification);
}
public CompletableFuture<List<String>>
listPersistentTopicsAsync(NamespaceName ns) {
String path = MANAGED_LEDGER_PATH + "/" + ns + "/persistent";
- return store.getChildren(path).thenApply(children ->
- children.stream().map(c ->
TopicName.get(TopicDomain.persistent.toString(), ns, decode(c)).toString())
- .collect(Collectors.toList())
- );
+ CompletableFuture<List<String>> childrenOfNamespace =
store.getChildren(path);
+ CompletableFuture<List<String>> topics =
childrenOfNamespace.thenCompose(children ->
+ !children.isEmpty() &&
children.get(0).startsWith(TopicName.BUCKET_ID_PERFIX)
+ ? children.stream().map(bucket ->
store.getChildren(path + "/" + bucket))
+ .collect(FutureUtil.toList())
+ : CompletableFuture.completedFuture(children));
Review Comment:
Nice catch. I did not realize this. I extracted the common part into a
method.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
