Replace custom Future implementations by CompletableFuture

[ywelsch]

This PR replaces our custom future implementations by Java 8's CompletableFuture, which results in less custom code to maintain (see PR stats), and enables the use of more powerful async programming features across our codebase. Furthermore, it fixes an issue with CompletableFuture, where it's not properly bubbling up Errors to the uncaught exception handler.

[elasticmachine]

Pinging @elastic/es-core-infra

[bleskes]

Looks awesome. I left minor comments.

[bleskes]

old code caught exceptions here and routed them to listener.onFailure()

[ywelsch]

Good catch. I'm not a fan of having this anti-pattern in library code but I'm still making this change for BWC purposes. In the future (pun intended) we can also think about adding these methods directly to BaseFuture as every Future has now listening capabilities.

[bleskes]

💯

[bleskes]

same comment about exception handling.

[Tim-Brooks]

I have a comment. This is similar to work I did in org.elasticsearch.common.concurrent.CompletableContext. One thing that I have noticed recently is that CompletableFuture catches Throwable that occurs in listeners. So:

CompletableContext<Object> context = new CompletableContext<>();
context.addListener((o, e) -> {
        assert false : "Should not fail";
});
context.complete(null);

That assertion is swallowed and never seen again. This is because when the listener is executed, CompletableFuture catches it and completes the other CompletableFuture that was created when the listener was attached. You have to manually interrogate that future to find it.

    final boolean uniWhenComplete(Object r,
                                  BiConsumer<? super T,? super Throwable> f,
                                  UniWhenComplete<T> c) {
        T t; Throwable x = null;
        if (result == null) {
            try {
                if (c != null && !c.claim())
                    return false;
                if (r instanceof AltResult) {
                    x = ((AltResult)r).ex;
                    t = null;
                } else {
                    @SuppressWarnings("unchecked") T tr = (T) r;
                    t = tr;
                }
                f.accept(t, x);
                if (x == null) {
                    internalComplete(r);
                    return true;
                }
            } catch (Throwable ex) {
                if (x == null)
                    x = ex;
                else if (x != ex)
                    x.addSuppressed(ex);
            }
            completeThrowable(x, r);
        }
        return true;
    }

I had not decided how I was going to fix that issue. But I was thinking of wrapping attached listeners, catching Error and doing the throw it to another thread thing.

I just thought I would mention that here as I assume it will impact this work and it was causing me issues for assertions to disappear.

[jasontedor]

Note that we handled a similar problem in #28667.

[Tim-Brooks]

The recent updates seem like they will work. The only downside seems to be that all the CompletionStage Apis (like whenComplete) take Throwable. People that use the BaseFuture will still have to deal with Throwable. But I think that is unavoidable without essentially creating our own CompletionStage interface that does not take Throwable.

[ywelsch]

I have reworked this PR so that the CompletableFutures now properly bubble up Errors to the uncaught exception handler.

People that use the BaseFuture will still have to deal with Throwable.

yes, but only if they make use of the methods on CompletionStage that require to do so. We can also provide convenient alternatives on BaseFuture if need be.

As follow-ups, I would like to:

• rename BaseFuture to something like just ESFuture
• remove some of the subclasses of BaseFuture, as the functionality is now directly available in the super class (e.g. ListenableFuture).

[elasticsearchmachine]

Hi @ywelsch, I've created a changelog YAML for you.

[elasticsearchmachine]

Pinging @elastic/es-core-infra (Team:Core/Infra)