@Test public void testSupplyAsyncFailure() throws Exception { final String exceptionMessage = "Test exception"; final FlinkException testException = new FlinkException(exceptionMessage); final CompletableFuture<Object> future = FutureUtils.supplyAsync( () -> { throw testException; }, TestingUtils.defaultExecutor()); try { future.get(); fail("Expected an exception."); } catch (ExecutionException e) { assertThat(ExceptionUtils.findThrowableWithMessage(e, exceptionMessage).isPresent(), is(true)); } }
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); }
FutureUtils { 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); } }
FutureUtils { 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); }  }
FutureUtils { 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); }  static CompletableFuture<T> retry(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Executor executor); static CompletableFuture<T> retryWithDelay(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Time retryDelay,
			final Predicate<Throwable> retryPredicate,
			final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> retryWithDelay(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Time retryDelay,
			final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> retrySuccessfulWithDelay(
		final Supplier<CompletableFuture<T>> operation,
		final Time retryDelay,
		final Deadline deadline,
		final Predicate<T> acceptancePredicate,
		final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit timeUnit); static CompletableFuture<T> orTimeout(
		CompletableFuture<T> future,
		long timeout,
		TimeUnit timeUnit,
		Executor timeoutFailExecutor); static T runIfNotDoneAndGet(RunnableFuture<T> future); static CompletableFuture<Void> runAfterwards(CompletableFuture<?> future, RunnableWithException runnable); static CompletableFuture<Void> runAfterwardsAsync(CompletableFuture<?> future, RunnableWithException runnable); static CompletableFuture<Void> runAfterwardsAsync(
		CompletableFuture<?> future,
		RunnableWithException runnable,
		Executor executor); static CompletableFuture<Void> composeAfterwards(
			CompletableFuture<?> future,
			Supplier<CompletableFuture<?>> composedAction); static ConjunctFuture<Collection<T>> combineAll(Collection<? extends CompletableFuture<? extends T>> futures); static ConjunctFuture<Void> waitForAll(Collection<? extends CompletableFuture<?>> futures); static ConjunctFuture<Void> completeAll(Collection<? extends CompletableFuture<?>> futuresToComplete); static CompletableFuture<T> completedExceptionally(Throwable cause); static CompletableFuture<T> supplyAsync(SupplierWithException<T, ?> supplier, Executor executor); static FiniteDuration toFiniteDuration(Time time); static Time toTime(FiniteDuration finiteDuration); static CompletableFuture<T> toJava(Future<U> scalaFuture); static CompletableFuture<OUT> thenApplyAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Function<? super IN, ? extends OUT> applyFun); static CompletableFuture<OUT> thenComposeAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Function<? super IN, ? extends CompletionStage<OUT>> composeFun); static CompletableFuture<IN> whenCompleteAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		BiConsumer<? super IN, ? super Throwable> whenCompleteFun); static CompletableFuture<Void> thenAcceptAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Consumer<? super IN> consumer); static CompletableFuture<OUT> handleAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		BiFunction<? super IN, Throwable, ? extends OUT> handler); static void assertNoException(CompletableFuture<?> completableFuture); static void handleUncaughtException(CompletableFuture<?> completableFuture, Thread.UncaughtExceptionHandler uncaughtExceptionHandler); }
FutureUtils { 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); }  static CompletableFuture<T> retry(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Executor executor); static CompletableFuture<T> retryWithDelay(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Time retryDelay,
			final Predicate<Throwable> retryPredicate,
			final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> retryWithDelay(
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Time retryDelay,
			final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> retrySuccessfulWithDelay(
		final Supplier<CompletableFuture<T>> operation,
		final Time retryDelay,
		final Deadline deadline,
		final Predicate<T> acceptancePredicate,
		final ScheduledExecutor scheduledExecutor); static CompletableFuture<T> orTimeout(CompletableFuture<T> future, long timeout, TimeUnit timeUnit); static CompletableFuture<T> orTimeout(
		CompletableFuture<T> future,
		long timeout,
		TimeUnit timeUnit,
		Executor timeoutFailExecutor); static T runIfNotDoneAndGet(RunnableFuture<T> future); static CompletableFuture<Void> runAfterwards(CompletableFuture<?> future, RunnableWithException runnable); static CompletableFuture<Void> runAfterwardsAsync(CompletableFuture<?> future, RunnableWithException runnable); static CompletableFuture<Void> runAfterwardsAsync(
		CompletableFuture<?> future,
		RunnableWithException runnable,
		Executor executor); static CompletableFuture<Void> composeAfterwards(
			CompletableFuture<?> future,
			Supplier<CompletableFuture<?>> composedAction); static ConjunctFuture<Collection<T>> combineAll(Collection<? extends CompletableFuture<? extends T>> futures); static ConjunctFuture<Void> waitForAll(Collection<? extends CompletableFuture<?>> futures); static ConjunctFuture<Void> completeAll(Collection<? extends CompletableFuture<?>> futuresToComplete); static CompletableFuture<T> completedExceptionally(Throwable cause); static CompletableFuture<T> supplyAsync(SupplierWithException<T, ?> supplier, Executor executor); static FiniteDuration toFiniteDuration(Time time); static Time toTime(FiniteDuration finiteDuration); static CompletableFuture<T> toJava(Future<U> scalaFuture); static CompletableFuture<OUT> thenApplyAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Function<? super IN, ? extends OUT> applyFun); static CompletableFuture<OUT> thenComposeAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Function<? super IN, ? extends CompletionStage<OUT>> composeFun); static CompletableFuture<IN> whenCompleteAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		BiConsumer<? super IN, ? super Throwable> whenCompleteFun); static CompletableFuture<Void> thenAcceptAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		Consumer<? super IN> consumer); static CompletableFuture<OUT> handleAsyncIfNotDone(
		CompletableFuture<IN> completableFuture,
		Executor executor,
		BiFunction<? super IN, Throwable, ? extends OUT> handler); static void assertNoException(CompletableFuture<?> completableFuture); static void handleUncaughtException(CompletableFuture<?> completableFuture, Thread.UncaughtExceptionHandler uncaughtExceptionHandler);  }