@Test
public void testRetryCancellation() throws Exception {

	final int retries = 10;
	final AtomicInteger atomicInteger = new AtomicInteger(0);
	final OneShotLatch notificationLatch = new OneShotLatch();
	final OneShotLatch waitLatch = new OneShotLatch();
	final AtomicReference<Throwable> atomicThrowable = new AtomicReference<>(null);
	CompletableFuture<?> retryFuture = FutureUtils.retry(
		() ->
			CompletableFuture.supplyAsync(
				() -> {
					if (atomicInteger.incrementAndGet() == 2) {
						notificationLatch.trigger();
						try {
							waitLatch.await();
						} catch (InterruptedException e) {
							atomicThrowable.compareAndSet(null, e);
						}
					}
					throw new CompletionException(new FlinkException("Test exception"));
				},
				TestingUtils.defaultExecutor()),
		retries,
		TestingUtils.defaultExecutor());
	// await that we have failed once
	notificationLatch.await();
	assertFalse(retryFuture.isDone());
	// cancel the retry future
	retryFuture.cancel(false);
	// let the retry operation continue
	waitLatch.trigger();
	assertTrue(retryFuture.isCancelled());
	assertEquals(2, atomicInteger.get());
	if (atomicThrowable.get() != null) {
		throw new FlinkException("Exception occurred in the retry operation.", atomicThrowable.get());
	}
}