Unable to understand this CompletableFuture behaviour

[Ranagal Adapla]

Hello forum veterans,

I need a small help in understanding this CompletableFuture. I am learning this topic for my understanding.

As you can see, I first generate some random integer list. And then I try to find out max, min and near to 100 element in that list. And these three tasks should happen in parallel.

public static void delay(int sec) {
		try {
			Thread.sleep(sec);
		} catch (InterruptedException e) {
			System.out.println("Exception Occurred: "+LocalTime.now()+" : "+e);
		}
	}

public static Integer findMax(List<Integer> list) {
		System.out.println("findMax: "+Thread.currentThread().getName()+" : "+LocalDateTime.now());
		delay(1);
		return list.stream().max(Integer::compare).get();
	}
	
	public static Integer findMin(List<Integer> list) {
		System.out.println("findMin: "+Thread.currentThread().getName()+" : "+LocalDateTime.now());
		return list.stream().min(Integer::compare).get();
	}
	
	public static Integer findNear100(List<Integer> list) {
		System.out.println("findNear100: "+Thread.currentThread().getName()+" : "+LocalDateTime.now());
		int distance = Integer.MAX_VALUE;
		int selectedInt = list.get(0);
		for(Integer item : list) {
			int currentDistance = Math.abs(item - 100);
			if(currentDistance < distance) {
				distance = currentDistance;
				selectedInt = item;
			}
		}
		return selectedInt;
	}
	
	public static void main(String[] args) throws InterruptedException, ExecutionException {
		System.out.println("In main, Start: "+LocalTime.now());
		CompletableFuture<List<Integer>> cfWithList = CompletableFuture
												.supplyAsync(TestCompletableFuture::generate);
												
		CompletableFuture<Void> ctMax = cfWithList
											.thenApplyAsync(TestCompletableFuture::findMax)
											.thenAccept(value -> {
												System.out.println("Found max of list: "+value);
											});
		
		CompletableFuture<Void> ctMin = cfWithList
											.thenApplyAsync(TestCompletableFuture::findMin)
											.thenAccept(value -> {
												System.out.println("Found min of list: "+value);
											});
		
		CompletableFuture<Void> ctFindNear100 = cfWithList
													.thenApplyAsync(TestCompletableFuture::findNear100)
													.thenAccept(value -> {
														System.out.println("Found the nearest of 100: "+value);
													});
		
		CompletableFuture<Void> allCf = CompletableFuture.allOf(ctMax, ctMin, ctFindNear100);
		
		System.out.println("In main,   End: "+LocalTime.now());
	}

Sample Output:

In main, Start: 12:25:47.949939800
Generate: start, thread name:  ForkJoinPool.commonPool-worker-1
Generated List: [63, 55, 74, 76, 86, 35, 13, 33, 73, 29]
Generate: stop
findMax: ForkJoinPool.commonPool-worker-1 : 2022-11-26T12:25:47.973199
findMin: ForkJoinPool.commonPool-worker-2 : 2022-11-26T12:25:47.974763600
findNear100: ForkJoinPool.commonPool-worker-3 : 2022-11-26T12:25:47.975773900
Found the nearest of 100: 86
In main,   End: 12:25:47.975773900

Question: Why are other methods not printing anything ?

My observation: If I remove the delay, it simply works fine like below: But why ?

In main, Start: 12:30:01.922193700
Generate: start, thread name:  ForkJoinPool.commonPool-worker-1
Generated List: [79, 33, 63, 19, 20, 73, 35, 28, 55, 85]
Generate: stop
findMax: ForkJoinPool.commonPool-worker-1 : 2022-11-26T12:30:01.944750800
findMin: ForkJoinPool.commonPool-worker-2 : 2022-11-26T12:30:01.946752400
findNear100: ForkJoinPool.commonPool-worker-3 : 2022-11-26T12:30:01.946752400
Found the nearest of 100: 85
In main,   End: 12:30:01.947536400
Found min of list: 19
Found max of list: 85


Regards,
Ranagal

[Rob Spoor]

Because you don't provide an Executor when calling supplyAsync, the commons fork-join pool is used for the work performed in the completable futures. That contains so-called daemon threads. Whereas non-daemon threads (like main) must end before the JVM ends, daemon threads will be "cut off" when the JVM ends (they don't prevent the JVM from exiting). In other words - the main method doesn't wait until the completable futures are actually done.

There are two ways to handle this:
1) Use an explicit ExecutorService, usually created by calling on of the static methods of Executors. While the ExecutorService hasn't been shut down by an explicit call, it will prevent the JVM from exiting.
2) Call join() or .get() on the last completable future. That will make the current (main) thread wait until the completable future is done.

[Ranagal Adapla]

Oh Ok. Thanks for clearing my doubts.

[Rob Spoor]

You're welcome.

[Stephan van Hulst]

Please note that your delay() method takes seconds, but it's implemented wrongly because Thread.sleep() takes milliseconds.

[Ranagal Adapla]

Stephan van Hulst wrote:Please note that your delay() method takes seconds, but it's implemented wrongly because Thread.sleep() takes milliseconds.

I in fact had something like sec * 1000.

But it was one second, and then I removed 1000 and just kept sec so that I can pass just 1 when I am calling delay for my testing. Even then that method was not printing anything on the console. So, my attention went to that part and while posting the question here, I forgot to add back that 1000. This is how I missed it.

Anyway, thanks for catching it.

[Himai Minh]

Hi, I test your code and found out that regardless delay is there or not, the main thread won't wait for the three completablefuture calls.
Here is what I do to fix it. Each completablefuture calls the join method. They will all finish before the main thread.
Add the following code above the System.out.println("In main, End: "+ LocalTime.now());

CompletableFuture<Void> allCf = CompletableFuture.allOf(ctMax, ctMin, ctFindNear100) ;
            List<CompletableFuture<Void>> cflist = List.of(ctMax, ctMin, ctFindNear100);
            allCf.
                thenApply(v -> cflist
                        .stream()
                        .map(CompletableFuture::join)
                        .collect(Collectors.toList())).join();
             System.out.println("In main,   End: "+LocalTime.now());

[Himai Minh]

Also, another workaround is :

 public static void main(String[] args) throws InterruptedException, ExecutionException {
         System.out.println("In main, Start: "+LocalTime.now());
        
         executorService.submit(()-> {
                               List<Integer> input = generate();
                               var value= findMax(input);
                               System.out.println("Found max of list: "+value);
                            });

         executorService.submit(()-> {
                                     List<Integer> input = generate();
                                     var value= findMin(input);
                                     System.out.println("Found min of list: "+value);
                                   }) ;

          executorService.submit(()-> {
                                      List<Integer> input = generate();
                                       var value= findNear100(input);
                                       System.out.println("Found the nearest of 100: "+value);
                                     });

   
        executorService.shutdown();
       System.out.println("In main,   End: "+LocalTime.now());
       }

[Ranagal Adapla]

Thank you for your time on this and also for the workaround ideas. This helped me a lot.

Regards,
Ranagal