heesung-sn commented on code in PR #20666:
URL: https://github.com/apache/pulsar/pull/20666#discussion_r1246064145
##########
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:
afaik, I think this is the current behavior in the modular load manager. I
think we can add a todo or a backlog issue to follow up.
--
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]
