### Question:
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> retrySuccesfulWithDelay(
		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<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<T> scalaFuture);  }### Answer:
@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)); } }
@Test public void testSupplyAsync() throws Exception { final CompletableFuture<Acknowledge> future = FutureUtils.supplyAsync( Acknowledge::get, TestingUtils.defaultExecutor()); assertThat(future.get(), is(Acknowledge.get())); }