Demogorgon314 commented on code in PR #20666:
URL: https://github.com/apache/pulsar/pull/20666#discussion_r1246248499
##########
pulsar-broker/src/main/java/org/apache/pulsar/broker/loadbalance/extensions/ExtensibleLoadManagerImpl.java:
##########
@@ -489,24 +489,27 @@ public CompletableFuture<Optional<String>>
selectAsync(ServiceUnitId bundle,
// Filter out brokers that do not meet the rules.
List<BrokerFilter> filterPipeline =
getBrokerFilterPipeline();
+ ArrayList<CompletableFuture<Map<String,
BrokerLookupData>>> futures =
+ new ArrayList<>(filterPipeline.size());
for (final BrokerFilter filter : filterPipeline) {
- try {
- filter.filter(availableBrokerCandidates, bundle,
context);
- // Preserve the filter successes result.
-
availableBrokers.keySet().retainAll(availableBrokerCandidates.keySet());
- } catch (BrokerFilterException e) {
- // TODO: We may need to revisit this error case.
- log.error("Failed to filter out brokers.", e);
- availableBrokerCandidates = new
HashMap<>(availableBrokers);
- }
- }
- if (availableBrokerCandidates.isEmpty()) {
- return
CompletableFuture.completedFuture(Optional.empty());
+ CompletableFuture<Map<String, BrokerLookupData>>
future =
+ filter.filter(availableBrokerCandidates,
bundle, context);
+ futures.add(future);
}
- Set<String> candidateBrokers =
availableBrokerCandidates.keySet();
+ CompletableFuture<Optional<String>> result = new
CompletableFuture<>();
+ FutureUtil.waitForAll(futures).whenComplete((__, ex) -> {
+ if (ex != null) {
+ log.error("Failed to filter out brokers when
select bundle: {}", bundle, ex);
Review Comment:
Yes, I added todo comment here.
--
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]
