public static <T> void forward(CompletableFuture<T> source, CompletableFuture<T> target) {
        source.whenComplete(forwardTo(target));
    }
@Test
    void testForwardExceptionally() {
        final CompletableFuture<String> source = new CompletableFuture<>();
        final CompletableFuture<String> target = new CompletableFuture<>();

        FutureUtils.forward(source, target);

        assertThat(source).isNotDone();
        assertThat(target).isNotDone();

        final FlinkException expectedCause = new FlinkException("Expected exception");
        source.completeExceptionally(expectedCause);

        assertThat(source).isDone();
        assertThat(target).isDone();

        assertThatFuture(source)
                .eventuallyFailsWith(ExecutionException.class)
                .extracting(Throwable::getCause)
                .isEqualTo(expectedCause);
        assertThatFuture(target)
                .eventuallyFailsWith(ExecutionException.class)
                .extracting(Throwable::getCause)
                .isEqualTo(expectedCause);
    }