navina commented on a change in pull request #8272:
URL: https://github.com/apache/pinot/pull/8272#discussion_r817385408
##########
File path:
pinot-core/src/main/java/org/apache/pinot/core/query/reduce/StreamingReduceService.java
##########
@@ -106,15 +108,17 @@ public BrokerResponseNative
reduceOnStreamResponse(BrokerRequest brokerRequest,
return brokerResponseNative;
}
- private static void processIterativeServerResponse(StreamingReducer reducer,
ExecutorService executorService,
+ @VisibleForTesting
+ static void processIterativeServerResponse(StreamingReducer reducer,
ExecutorService executorService,
Map<ServerRoutingInstance, Iterator<Server.ServerResponse>>
serverResponseMap, long reduceTimeOutMs,
- ExecutionStatsAggregator aggregator) throws Exception {
+ ExecutionStatsAggregator aggregator) throws ExecutionException {
int cnt = 0;
- Future[] futures = new Future[serverResponseMap.size()];
- CountDownLatch countDownLatch = new
CountDownLatch(serverResponseMap.size());
-
+ Future<Void>[] futures = new Future[serverResponseMap.size()];
+ // based on ideas from on
https://stackoverflow.com/questions/19348248/waiting-on-a-list-of-future
+ // mostly because we need to handle and transfer the exception in threads
into caller thread.
+ ExecutorCompletionService<Void> countDownHelper = new
ExecutorCompletionService<>(executorService);
Review comment:
wouldn't `CompletableFuture.allOf` satisfy your requirement here -
https://docs.oracle.com/javase/8/docs/api/java/util/concurrent/CompletableFuture.html#allOf-java.util.concurrent.CompletableFuture...-
?
Or is it too invasive change to replace `Future` with `CompletableFuture`?
--
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...@pinot.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
