Java multiple completablefuture

Question

I wrote the following snipet code within Cumulocity platform.

runForTenant

must be in a separate thread. What is your an opinion about my code. is there any need for optimization? Thanks

LinkedList<CompletableFuture<Void>> asyncThreads = new LinkedList<CompletableFuture<Void>>();
 CompletableFuture<Iterable<ManagedObjectRepresentation>> response = new CompletableFuture<>();
 subscriptions.runForEachTenant(() -> {
 String tenant = subscriptions.getTenant();
 asyncThreads.add(CompletableFuture.runAsync(() -> {
 subscriptions.runForTenant(tenant, () -> {
 Iterable<ManagedObjectRepresentation> objects = inventoryApi
 .getManagedObjectsByFilter(customInventoryFilter).get().allPages();
 if (!ObjectUtils.isEmpty(objects)) {
 response.complete(objects);
 }
 });
 })); // end of runAsync()
 });
 boolean allThreadsAreDone = true;
 do { // until all threads are done ...
 try {
 return response.get(50, TimeUnit.MILLISECONDS); // wait for first result
 } catch (TimeoutException e) {
 // device with UUID Not found in time
 // check all threads are done/closed/stopped in all fashions
 allThreadsAreDone = true;
 for (int i = 0; allThreadsAreDone && i < asyncThreads.size(); i++) {
 allThreadsAreDone = asyncThreads.get(i).isDone();
 }
 } catch (ExecutionException execExc) {
 logger.error("Error on lookup: " + execExc.getLocalizedMessage());
 } catch (InterruptedException interExc) {
 logger.error("Error on lookup: " + interExc.getLocalizedMessage());
 }
 } while (!allThreadsAreDone);

I want to optimize the last path of the code:

boolean allThreadsAreDone = true;
 do { // until all threads are done ...
 try {
 return response.get(50, TimeUnit.MILLISECONDS); // wait for first result
 } catch (TimeoutException e) {
 // device with UUID Not found in time
 // check all threads are done/closed/stopped in all fashions
 allThreadsAreDone = true;
 for (int i = 0; allThreadsAreDone && i < asyncThreads.size(); i++) {
 allThreadsAreDone = asyncThreads.get(i).isDone();
 }
 } catch (ExecutionException execExc) {
 logger.error("Error on lookup: " + execExc.getLocalizedMessage());
 } catch (InterruptedException interExc) {
 logger.error("Error on lookup: " + interExc.getLocalizedMessage());
 }
 } while (!allThreadsAreDone);

• Andrey Lebedenko: What does subscriptions::runForEachTenant do? Executes a Runnable? Naming suggests it should execute a Function or a Consumer of tenant.
• Fortran: @AndreyLebedenko yes, it executes Runnable

Answer 1

I have one advice, but since the code is not compiling, it's a bit hard to make a proper review.

Instead of using an java.util.LinkedList, I suggest that you use the java.util.concurrent.ExecutorService to handle the threads and use it to check the end of the threads.

 ExecutorService taskExecutor = Executors.newFixedThreadPool(4);
 CompletableFuture<Iterable<ManagedObjectRepresentation>> response = new CompletableFuture<>();
 subscriptions.runForEachTenant(() -> {
 String tenant = subscriptions.getTenant();
 taskExecutor.execute(CompletableFuture.runAsync(() -> {
 subscriptions.runForTenant(tenant, () -> {
 Iterable<ManagedObjectRepresentation> objects = inventoryApi
 .getManagedObjectsByFilter(customInventoryFilter).get().allPages();
 if (!ObjectUtils.isEmpty(objects)) {
 response.complete(objects);
 }
 });
 }));
 });
 taskExecutor.shutdown();
 try {
 taskExecutor.awaitTermination(Long.MAX_VALUE, TimeUnit.NANOSECONDS);
 } catch (InterruptedException e) {
 }
 // The threads are finished.

Ref: https://stackoverflow.com/questions/1250643/how-to-wait-for-all-threads-to-finish-using-executorservice

• dustytrash: If the code does not compile it's off topic and shouldn't be answered
• Fortran: it is not possible to show you the whole code ;)
• Andrey Lebedenko: Calling taskExecutor.execute(CompletableFuture.runAsync( does not make much sense, since CompletableFuture.runAsync in this case we use ForkJoin internally. Instead pass an executor as a second parameter of CompletableFuture.runAsync itself. The number of threads in executor is also an open question. Also, in case of multiple parallel calls to this block, it may quickly end up in resource starvation, due to multiple executors competing for processor. Therefore it is better to pass an executor as a parameter from outside of the block, and don't shut it down.
• Fortran: Also taskExecutor.execute requires Runnable docs.oracle.com/javase/7/docs/api/java/util/concurrent/…

Answer 2

The second loop do...while can be replaced by

 asyncThreads.forEach(cf -> cf.join()); // blocks until all CompletableFutures join in any order.

Though those are not threads, per say, they are CompletableFutures.