public static <T> CompletableFuture<T> supplyAsync(
            SupplierWithException<T, ?> supplier, Executor executor) {
        return CompletableFuture.supplyAsync(
                () -> {
                    try {
                        return supplier.get();
                    } catch (Throwable e) {
                        throw new CompletionException(e);
                    }
                },
                executor);
    }
@Test
    void testSupplyAsyncFailure() {
        final String exceptionMessage = "Test exception";
        final FlinkException testException = new FlinkException(exceptionMessage);
        final CompletableFuture<Object> future =
                FutureUtils.supplyAsync(
                        () -> {
                            throw testException;
                        },
                        EXECUTOR_RESOURCE.getExecutor());

        assertThatFuture(future)
                .eventuallyFailsWith(ExecutionException.class)
                .withCause(testException);
    }