### Question:
FutureUtils { public static <T> CompletableFuture<T> retryWithDelay( final Supplier<CompletableFuture<T>> operation, final int retries, final Time retryDelay, final Predicate<Throwable> retryPredicate, final ScheduledExecutor scheduledExecutor) { final CompletableFuture<T> resultFuture = new CompletableFuture<>(); retryOperationWithDelay( resultFuture, operation, retries, retryDelay, retryPredicate, scheduledExecutor); return resultFuture; }  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 testRetryWithDelay() throws Exception { final int retries = 4; final Time delay = Time.milliseconds(5L); final AtomicInteger countDown = new AtomicInteger(retries); long start = System.currentTimeMillis(); CompletableFuture<Boolean> retryFuture = FutureUtils.retryWithDelay( () -> { if (countDown.getAndDecrement() == 0) { return CompletableFuture.completedFuture(true); } else { return FutureUtils.completedExceptionally(new FlinkException("Test exception.")); } }, retries, delay, TestingUtils.defaultScheduledExecutor()); Boolean result = retryFuture.get(); long completionTime = System.currentTimeMillis() - start; assertTrue(result); assertTrue("The completion time should be at least rertries times delay between retries.", completionTime >= retries * delay.toMilliseconds()); }