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);

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

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.