private static <T> void retryOperationWithDelay(final CompletableFuture<T> resultFuture,
      final Supplier<CompletableFuture<T>> operation, final int retries, Duration retryDelay,
      final Predicate<Throwable> retryPredicate, Function<Throwable, Integer> retryDelayFromException,
      final ScheduledExecutorService scheduler) {

    if (!resultFuture.isDone()) {
      final CompletableFuture<T> operationResultFuture = operation.get();

      operationResultFuture.whenComplete((t, throwable) -> {
        if (throwable != null) {
          if (throwable instanceof CancellationException) {
            resultFuture.completeExceptionally(throwable);
          } else {
            throwable = extractFutureExceptionCause(throwable);
            if (!retryPredicate.test(throwable)) {
              resultFuture.completeExceptionally(throwable);
            } else if (retries > 0) {
              long retryDelayFromExceptionMs = retryDelayFromException.apply(throwable);
              long retryDelayMillis = retryDelayFromExceptionMs > 0 ? retryDelayFromExceptionMs : retryDelay.toMillis();
              if (scheduler == null) {
                if (retryDelayMillis > 0) {
                  logger.warn("No scheduler for delay execution. waitTimeMs: {}", retryDelayMillis);
                  retryOperationWithDelay(resultFuture, operation, retries - 1, retryDelay, retryPredicate,
                      retryDelayFromException, scheduler);
                }
              } else {
                final ScheduledFuture<?> scheduledFuture = scheduler.schedule(
                    () -> retryOperationWithDelay(resultFuture, operation, retries - 1, retryDelay, retryPredicate,
                        retryDelayFromException, scheduler), retryDelayMillis, MILLISECONDS);

                resultFuture.whenComplete((innerT, innerThrowable) -> scheduledFuture.cancel(false));
              }
            } else {
              resultFuture.completeExceptionally(throwable);
            }
          }
        } else {
          resultFuture.complete(t);
        }
      });
    }
  }
@param <T> type of the future's result
@param resultFuture to complete
@param operation to retry
@param retries until giving up
@param retryDelay duration of retries
@param retryPredicate Predicate to test whether an exception is retryable
@param retryDelayFromException function to get retry delay from the exception if present
@param scheduler executor to schedule retries
ambry-commons/src/main/java/com/github/ambry/commons/FutureUtils.java