BewareMyPower commented on code in PR #23049:
URL: https://github.com/apache/pulsar/pull/23049#discussion_r1682388030
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/namespace/NamespaceService.java:
##########
@@ -1509,6 +1514,31 @@ public CompletableFuture<List<String>>
getListOfTopics(NamespaceName namespaceNa
}
}
+ public CompletableFuture<List<String>> getListOfUserTopics(NamespaceName
namespaceName, Mode mode) {
+ String key = String.format("%s://%s", mode, namespaceName);
+ CompletableFuture<List<String>> queryRes =
+ inProgressQueryUserTopics.computeIfAbsent(key, k -> {
+ CompletableFuture<List<String>> res = new CompletableFuture<>();
+ // Switch thread to avoid blocking other threads who are calling
the current method.
+ pulsar.getExecutor().execute(() -> {
+ getListOfTopics(namespaceName, mode).thenApply(list -> {
+ return TopicList.filterSystemTopic(list);
+ }).whenComplete((topics, ex) -> {
+ if (ex != null) {
+ res.completeExceptionally(ex);
+ } else {
+ res.complete(topics);
+ }
+ });
+ });
+ return res;
+ });
+ queryRes.whenComplete((ignore, ex) -> {
Review Comment:
> the CompletableFuture in the second code never complete.
See the following code and outputs.
```java
static class Cache {
private final Map<Integer, CompletableFuture<Integer>> futures = new
ConcurrentHashMap<>();
public CompletableFuture<Integer> run(int x) {
final var inserted = new MutableBoolean(false);
final var future = futures.computeIfAbsent(x, __ -> {
inserted.setTrue();
final var innerFuture = new CompletableFuture<Integer>();
CompletableFuture.runAsync(() -> {
try {
Thread.sleep(100);
} catch (InterruptedException e) {
}
innerFuture.complete(100);
});
return innerFuture;
});
if (inserted.isTrue()) {
future.whenComplete((__, ___) -> {
System.out.println("Remove " + x);
futures.remove(x, future);
});
}
return future;
}
}
public static void main(String[] args) throws ExecutionException,
InterruptedException {
final var cache = new Cache();
cache.run(1).thenAccept(System.out::println);
cache.run(1).thenAccept(System.out::println);
cache.run(1).thenAccept(System.out::println);
cache.run(1).thenAccept(System.out::println).get();
```
Outputs:
```
100
100
100
100
Remove 1
```
--
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-unsubscr...@pulsar.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
