How to ensure that CompletableFuture is completely finished before execute a piece of code?

user_635162 0
Viewed 739

I have this piece of code that I have two questions.

  1. I'm not sure why I see TimeoutException as there's no blocking anywhere.
  2. What I'm trying to achieve with the Collector is I have a class that will go in to collect a bunch of metrics and after the CompletableFuture is completely done then I would execute Collector to release metrics. Is finally guaranteed that it will be executed last as I think .get() is supposed to be blocked until it's finished?
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.function.Function;
import java.util.function.Supplier;

public class FutureWithCollector
{
    public static void main(String[] args) throws ExecutionException, InterruptedException
    {
        Collector collector = new Collector();
        finalize(() -> query(collector), collector);
    }

    private static void finalize(Supplier<CompletableFuture<String>> submission, Collector collector) throws ExecutionException, InterruptedException
    {
        ExecutorService executorService = Executors.newFixedThreadPool(1);
        CompletableFuture<String> s = CompletableFuture.supplyAsync(submission, executorService).thenCompose(Function.identity());
        try {
            String result = s.get(1000, TimeUnit.MILLISECONDS);
            System.out.println(result);
        } catch (TimeoutException e) {
            e.printStackTrace();
        } finally {
            System.out.println(collector.getI());
        }
    }

    private static CompletableFuture<String> query(Collector collector)
    {
        CompletableFuture<String> future = new CompletableFuture<>();
        return future.thenApply(r -> {
            collector.collectStuff();
            return "Hello";
        });
    }

}

class Collector
{
    private volatile int i;

    public void collectStuff()
    {
        i++;
    }

    public int getI()
    {
        return i;
    }
}

Output

java.util.concurrent.TimeoutException
    at java.util.concurrent.CompletableFuture.timedGet(CompletableFuture.java:1771)
    at java.util.concurrent.CompletableFuture.get(CompletableFuture.java:1915)
    at FutureWithCollector.finalize(FutureWithCollector.java:23)
    at FutureWithCollector.main(FutureWithCollector.java:15)
0
completable-future java
3 Answers

It's rather trivial, just change one of your methods:

 private static CompletableFuture<String> query(Collector collector) {

    return CompletableFuture.supplyAsync(() -> {
        collector.collectStuff(); 
        return "hello";
    });

}

You are doing:

CompletableFuture<String> future = new CompletableFuture<>();

which is documented as:

Creates a new incomplete CompletableFuture.

Essentially, no one completes this CompletableFuture, so you will always get a timeout, no matter how big it is.

You can also change your code a bit. If you want to run something, say that explicitly:

private static CompletableFuture<Void> query(Collector collector) {
     return CompletableFuture.runAsync(collector::collectStuff);
}

Then please notice that collectStuff increments a volatile, but these increments are not atomic.

And you can always use join instead of get and not handle checked exceptions (granted there is no join that takes a timeout).

user_10593720

I'm not sure why I see TimeoutException as there's no blocking anywhere.

As Louis Wasserman noted, it comes from s.get(1000, TimeUnit.MILLISECONDS) which is a blocking call. (Sure, it is only blocking for at most 1 second. But it is still blocking.)

Is finally guaranteed that it will be executed last.

Yes ... but that's not the salient question, because ...

as I think .get() is supposed to be blocked until it's finished?

You are NOT using get(). You are using get with a timeout!

A call to get with a timeout is NOT guaranteed to block until the task corresponding to the future has finished. It actually waits until EITHER the task has finished OR the 1 second timeout expires. (Which ever happens first.)

If you want to be sure that the task has finished, don't call get with a timeout on the CompletableFuture.

You can't have it both ways1. Either you wait until the task is finished (which could take indefinitely long) or you wait with a timeout (and the task may not have finished).

1 - ... unless you have a fully operational time machine which allows you to go into the future to find out what the task's result is going to be, and then come back to the present to deliver that value.

user_1399850

When you write s.get(1000, TimeUnit.MILLISECONDS), the compiler warns you to handle TimeoutException. Which is maybe why you have added a basic catch block.

catch (TimeoutException e) {
   e.printStackTrace();
}

In such a case, it is always helpful to see what the javadoc tells for that method,

/**
 * Waits if necessary for at most the given time for the computation
 * to complete, and then retrieves its result, if available.
 *
 * @param timeout the maximum time to wait
 * @param unit the time unit of the timeout argument
 * @return the computed result
 * @throws CancellationException if the computation was cancelled
 * @throws ExecutionException if the computation threw an
 * exception
 * @throws InterruptedException if the current thread was interrupted
 * while waiting
 * @throws TimeoutException if the wait timed out
 */
V get(long timeout, TimeUnit unit)
    throws InterruptedException, ExecutionException, TimeoutException;

Now you got 2 more Exceptions to handle, Isn't your IDE or compiler complaining you to handle them as well?

Also, finally will get executed after get() either returns desired value from Future or throws Exception (Not until it comes out of the get() since it is a blocking call).

user_61403140
Related