flink-runtime/src/main/java/org/apache/flink/runtime/concurrent/FutureUtils.java
/**
	 * Helper method which retries the provided operation in case of a failure.
	 *
	 * @param resultFuture to complete
	 * @param operation to retry
	 * @param retries until giving up
	 * @param executor to run the futures
	 * @param <T> type of the future's result
	 */
Helper method which retries the provided operation in case of a failure.
Helper method which retries the provided operation in case of a failure .
private static <T> void retryOperation(
			final CompletableFuture<T> resultFuture,
			final Supplier<CompletableFuture<T>> operation,
			final int retries,
			final Executor executor) {
		if (!resultFuture.isDone()) {
			final CompletableFuture<T> operationFuture = operation.get();
			operationFuture.whenCompleteAsync(
				(t, throwable) -> {
					if (throwable != null) {
						if (throwable instanceof CancellationException) {
							resultFuture.completeExceptionally(new RetryException("Operation future was cancelled.", throwable));
						} else {
							if (retries > 0) {
								retryOperation(
									resultFuture,
									operation,
									retries - 1,
									executor);
							} else {
								resultFuture.completeExceptionally(new RetryException("Could not complete the operation. Number of retries " +
									"has been exhausted.", throwable));
							}
						}
					} else {
						resultFuture.complete(t);
					}
				},
				executor);
			resultFuture.whenComplete(
				(t, throwable) -> operationFuture.cancel(false));
		}
	}
private static < T > void retryOperation ( final CompletableFuture < T > resultFuture , final Supplier < CompletableFuture < T > > operation , final int retries , final Executor executor ) { if ( ! resultFuture . isDone ( ) ) { final CompletableFuture < T > operationFuture = operation . get ( ) ; operationFuture . whenCompleteAsync ( ( t , throwable ) -> { if ( throwable != null ) { if ( throwable instanceof CancellationException ) { resultFuture . completeExceptionally ( new RetryException ( "Operation future was cancelled." , throwable ) ) ; } else { if ( retries > 0 ) { retryOperation ( resultFuture , operation , retries - 1 , executor ) ; } else { resultFuture . completeExceptionally ( new RetryException ( "Could not complete the operation. Number of retries " + "has been exhausted." , throwable ) ) ; } } } else { resultFuture . complete ( t ) ; } } , executor ) ; resultFuture . whenComplete ( ( t , throwable ) -> operationFuture . cancel ( false ) ) ; } }
Helper method which retries the provided operation in case of a failure.
Helper method which retries the provided operation in case of a failure .