lhotari commented on code in PR #23094:
URL: https://github.com/apache/pulsar/pull/23094#discussion_r1696367524
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/admin/impl/ClustersBase.java:
##########
@@ -771,8 +782,13 @@ private CompletableFuture<Void>
filterAndUnloadMatchedNamespaceAsync(NamespaceIs
.map(tenant ->
adminClient.namespaces().getNamespacesAsync(tenant));
return FutureUtil.waitForAll(completableFutureStream)
.thenApply(namespaces -> {
- // if namespace match any policy regex, add it
to ns list to be unload.
+ // Filter namespaces that have current cluster
in their replication_clusters
+ // if namespace match any policy regex, add it
to ns list to be unloaded.
return namespaces.stream()
+ .filter(namespaceName ->
adminClient.namespaces()
+
.getPoliciesAsync(namespaceName)
+ .thenApply(policies ->
policies.replication_clusters.contains(cluster))
+ .join())
Review Comment:
> Can you suggest how I can make it async?
There's a real problem already with the existing code, even without
unloading calls. I've explained some of that in the previous comment in
https://github.com/apache/pulsar/pull/23094#issuecomment-2257496114 .
One of the problems is that all tenants and namespaces will be listed
concurrently at once, without any concurrency limits. That alone will cause
problems.
To fix the problem, the solution for making asynchronous calls will need
concurrency limits. I'd suggest introducing a dependency to
```
<dependency>
<groupId>com.spotify</groupId>
<artifactId>completable-futures</artifactId>
<version>0.3.6</version>
</dependency>
```
and using the
https://github.com/spotify/completable-futures/blob/master/src/main/java/com/spotify/futures/ConcurrencyReducer.java
class for controlling the concurrency.
This challenge is that this is a systemic problem at a higher level and
solving this problem in this PR might feel overly complex. However, it's
possible to handle it incrementally and refactor later.
For making the code asynchronous without blocking calls, composition is
needed by using `thenCompose`/`thenApply`.
In this case, it's not trivial, so it requires a bit more thought than usual
since the unlimited concurrency problem needs to also be solved.
--
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]
