@Test
public void testRetryWithDelay() throws Exception {

	final int retries = 4;
	final Time delay = Time.milliseconds(50L);
	final AtomicInteger countDown = new AtomicInteger(retries);
	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());
	long start = System.currentTimeMillis();
	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());
}