@Test
public void testRetrySuccess() throws Exception {

	final int retries = 10;
	final AtomicInteger atomicInteger = new AtomicInteger(0);
	CompletableFuture<Boolean> retryFuture = FutureUtils.retry(
		() ->
			CompletableFuture.supplyAsync(
				() -> {
					if (atomicInteger.incrementAndGet() == retries) {
						return true;
					} else {
						throw new CompletionException(new FlinkException("Test exception"));
					}
				},
				TestingUtils.defaultExecutor()),
		retries,
		TestingUtils.defaultExecutor());
	assertTrue(retryFuture.get());
	assertTrue(retries == atomicInteger.get());
}