private <T> void recursiveAsyncRetry(Consumer<Callback<T>> call, Predicate<Throwable> isRetriable, RetryPolicy policy, Callback<T> userCallback,
      int attempts) {
    call.accept((result, exception) -> {
      if (exception != null) {
        int currAttempts = attempts + 1;
        if (currAttempts < policy.maxAttempts() && isRetriable.test(exception)) {
          int waitTimeMs = policy.waitTimeMs(currAttempts);
          logger.info("{} of {} attempts failed, will keep retrying after a {} ms backoff. exception='{}'",
              currAttempts, policy.maxAttempts(), waitTimeMs, exception);
          //Don't need the scheduler if no need to wait for retry and show warn message if scheduler is null but wait time larger than 0;
          if (scheduler == null) {
            if (waitTimeMs > 0) {
              logger.warn("No scheduler for delay execution. waitTimeMs: {}", waitTimeMs);
            }
            recursiveAsyncRetry(call, isRetriable, policy, userCallback, currAttempts);
          } else {
            scheduler.schedule(() -> recursiveAsyncRetry(call, isRetriable, policy, userCallback, currAttempts),
                waitTimeMs, TimeUnit.MILLISECONDS);
          }
        } else {
          logger.info("{} of {} attempts failed, completing operation. See exception below:", currAttempts,
              policy.maxAttempts(), exception);
          userCallback.onCompletion(null, exception);
        }
      } else {
        userCallback.onCompletion(result, null);
      }
    });
  }
Recursively retrying on the task call with specific policy and user defined callback.
@param call The function call will be executed and retried with failure.
@param isRetriable user defined {@link Predicate} for retry or not when exception happen.
@param policy The {@link RetryPolicy} to schedule retrying.
@param userCallback The user defined {@link Callback} to be executed after all retry attempts failed.
@param attempts The number of retries has been attempted.
ambry-commons/src/main/java/com/github/ambry/commons/RetryExecutor.java