Executor service not waiting until all threads get completed to return the result

Question

I am using an Executor service to execute set of parallel tasks and I need one done something after all the tasks get completed. But in my below implementation it is not happening as expected and it is not waiting till all done. Please refer to the code snippet below.

public void doParallelStuff() throws InterruptedException {
        
        final int THREAD_COUNT = 5;
        ExecutorService executorService = Executors.newFixedThreadPool(THREAD_COUNT);
        Set<String> myIdSet = Set.of("123", "234", "345", "897", "893"); // actual set is having 20k+ IDs
        CountDownLatch countDownLatch = new CountDownLatch(myIdSet.size());

        myIdSet.forEach(id -> executorService.execute(() -> {
            // retrieve & do processing the entity related to id
            countDownLatch.countDown();
        }));

        countDownLatch.await(10L, TimeUnit.SECONDS);
        executorService.shutdown();

        // do some other stuff after all completed
        // (but this point is getting executed before the all tasks get completed)
    }

Answer 1

Lose the CountDownLatch

You don't need the CountDownLatch. An executor service provides that feature, built-in.

shutdown & awaitTermination in ExecutorService

The key is making two calls:

• ExecutorService#shutdown to stop the executor service from accepting more task submissions.
• ExecutorService#awaitTermination to block and wait for all submitted tasks to finish.

Here is the main chunk of code excerpted from a full example app posted further down.

First we generate some fake input. And we define a thread-safe list to hold Result objects, with each submitted task instantiated and collecting a fresh Result object.

Set < Integer > ids = IntStream.range( 1 , 20_001 ).boxed().collect( Collectors.toSet() );
List < Result > results = Collections.synchronizedList( new ArrayList <>() );

Next we define a collection to hold our pending tasks. We populate with many Task objects.

List < Task > tasks = new ArrayList <>( ids.size() );
for ( Integer id : ids ) tasks.add( new Task( id , results ) );

We create an executor service with a reasonable number of threads, as appropriate to our deployment machine.

ExecutorService executorService = Executors.newFixedThreadPool( 7 );

Next we define a collection to keep all the Future objects returned by the executor service when we submit all our tasks.

List < Future < Boolean > > futures = null;

Submit all the tasks, capturing the futures.

try { futures = executorService.invokeAll( tasks , 5 , TimeUnit.MINUTES ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); }

Make those two crucial calls in this subroutine, to invoke a shutdown and wait for tasks to complete.

this.shutdownAndAwaitTermination( executorService );  // Blocks here, until either (a) executor service completes all assigned tasks, or (b) time-out fires.

The key concept is that the code above blocks until all submitted tasks are done… or until the specified time-out fires.

Lastly, we can report on the work.

this.report( futures , results );

Complete code example

package work.basil.example.parallel;

import java.time.Instant;
import java.util.*;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;

public class App
{
    public static void main ( String[] args )
    {
        App app = new App();
        app.demo();
    }

    record Result( Integer id , UUID label , Instant whenCreated ) { }

    class Task implements Callable < Boolean >
    {
        private Integer id;
        private List < Result > results;

        public Task ( Integer id , List < Result > results )
        {
            this.id = id;
            this.results = results;
        }

        @Override
        public Boolean call ( )
        {
            try { Thread.sleep( ThreadLocalRandom.current().nextInt( 0 , 5 ) ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); }  // Pretend this method takes a while to do its work.
            Result result = new Result( this.id , UUID.randomUUID() , Instant.now() );
            this.results.add( result );
            return Boolean.TRUE;
        }
    }

    private void demo ( )
    {
        System.out.println( "INFO - Demo starting. " + Instant.now() );

        Set < Integer > ids = IntStream.range( 1 , 20_001 ).boxed().collect( Collectors.toSet() );
        List < Result > results = Collections.synchronizedList( new ArrayList <>() );

        List < Task > tasks = new ArrayList <>( ids.size() );
        for ( Integer id : ids ) tasks.add( new Task( id , results ) );

        ExecutorService executorService = Executors.newFixedThreadPool( 7 );
        List < Future < Boolean > > futures = null;
        try { futures = executorService.invokeAll( tasks , 5 , TimeUnit.MINUTES ); } catch ( InterruptedException e ) { throw new RuntimeException( e ); }
        this.shutdownAndAwaitTermination( executorService );  // Blocks here, until either (a) executor service completes all assigned tasks, or (b) time-out fires.
        this.report( futures , results );

        System.out.println( "INFO - Demo ending. " + Instant.now() );
    }

    private void report ( final List < Future < Boolean > > futures , final List < Result > results )
    {
        List < Future < Boolean > > failedFutures =
                futures.stream().filter( future -> {
                            try { return future.isCancelled() || ! future.isDone() || Objects.isNull( future.get() ) || Objects.equals( future.get() , Boolean.FALSE ); }
                            catch ( InterruptedException e ) { throw new RuntimeException( e ); }
                            catch ( ExecutionException e ) { throw new RuntimeException( e ); }
                        } )
                        .toList();
        System.out.println( failedFutures.size() + " failed futures:\n" + failedFutures );

        List < Future < Boolean > > doneFutures = futures.stream().filter( future -> {
            try { return future.isDone() && future.get().equals( Boolean.TRUE ); }
            catch ( InterruptedException e ) { throw new RuntimeException( e ); }
            catch ( ExecutionException e ) { throw new RuntimeException( e ); }
        } ).toList();
        System.out.println( doneFutures.size() + " done futures." );

        System.out.println( "First 10 results: " );
        results.stream().limit( 10 ).forEach( System.out :: println );
    }

    // Boilerplate taken from Javadoc of `ExecutorService`, and slightly edited.
    private void shutdownAndAwaitTermination ( ExecutorService executorService )
    {
        executorService.shutdown(); // Disable new tasks from being submitted
        try
        {
            // Wait a while for existing tasks to terminate
            if ( ! executorService.awaitTermination( 5 , TimeUnit.MINUTES ) )
            {
                executorService.shutdownNow(); // Cancel currently executing tasks
                // Wait a while for tasks to respond to being cancelled
                if ( ! executorService.awaitTermination( 5 , TimeUnit.MINUTES ) )
                { System.err.println( "Executor service failed to terminate." ); }
            }
        }
        catch ( InterruptedException ex )
        {
            executorService.shutdownNow();       // (Re-)Cancel if current thread also interrupted
            Thread.currentThread().interrupt();           // Preserve interrupt status
        }
    }
}

When run:

INFO - Demo starting. 2022-07-22T06:23:19.725428Z
0 failed futures:
[]
20000 done futures.
First 10 results: 
Result[id=7, label=609c5f42-3108-4ed7-a92d-0485016238c7, whenCreated=2022-07-22T06:23:19.765087Z]
Result[id=2, label=5dad4f85-e572-4a06-b226-f402fd3295c3, whenCreated=2022-07-22T06:23:19.765101Z]
Result[id=6, label=97d1786a-830b-42e5-97bb-48e6342fe3b1, whenCreated=2022-07-22T06:23:19.765140Z]
Result[id=4, label=17ae84ac-0dd8-477a-8b67-d4770643a015, whenCreated=2022-07-22T06:23:19.765191Z]
Result[id=5, label=55d4f4ba-9f5d-42f0-9af6-9e6c53159ce9, whenCreated=2022-07-22T06:23:19.765220Z]
Result[id=1, label=2d55c303-8366-441d-a62b-3168158329e0, whenCreated=2022-07-22T06:23:19.765261Z]
Result[id=3, label=ad4c7d04-7917-432e-8ad5-b0312b03d1bd, whenCreated=2022-07-22T06:23:19.765292Z]
Result[id=9, label=2774d965-f179-4b5b-a50f-361f5e4a1a08, whenCreated=2022-07-22T06:23:19.765346Z]
Result[id=13, label=54f48070-2967-4307-b4f9-80d34ece5332, whenCreated=2022-07-22T06:23:19.765378Z]
Result[id=11, label=00899822-591e-4af2-8b32-d496c3997e65, whenCreated=2022-07-22T06:23:19.766582Z]
INFO - Demo ending. 2022-07-22T06:23:27.027545Z

Parallel Streams

Using parallel streams may be a simpler route, depending on your specific needs. You have less control, such as not specifying the number of threads (by default), and less flexibility in the face of failures. But you get drastically shorter simpler code.

Define a record to hold each result. This is similar to above, but here we use a String type for id to be closer to your original Question.

record Result( String id , UUID label , Instant whenCreated ) { }

Create some bogus input data, as strings.

List < String > ids =
        IntStream
                .range( 1 , 20_001 )
                .boxed()
                .map( Object :: toString )
                .toUnmodifiableList();

And do the work, all in a single line of code. The .parallelStream automatically handles getting a pool of threads of a reasonable size, and then doing the work of instantiating our Result object as tasks assigned to the various threads.

List < Result > results =
        ids
                .parallelStream()
                .map( ( String id ) -> new Result( id , UUID.randomUUID() , Instant.now() ) )
                .toUnmodifiableList();

Lastly, report results.

results.stream().limit( 10 ).forEach( System.out :: println );

Be aware that using parallel streams only makes sense where you have a lengthy workload. For a brief workload, using parallel streams rather than single stream may actually take more time, as there is overhead in setting up, managing, and collecting results across threads.

Project Loom

Everything above is relevant for today's Java. If Project Loom succeeds, and it looks like it will, the rules of the game will be changed… for the better.

To learn more, see posts, interviews, and presentations by members of the Loom team such as Ron Pressler and Alan Bateman. Focus on the more recent info, as Loom has evolved over time. And see the Java JEPs for virtual threads, and for structured concurrency. Previews are found in Java 19.