public static <T> CompletableFuture<T> executeAsyncWithRetries(String opName,
      Supplier<? extends CompletionStage<T>> action, Predicate<? extends Throwable> abortRetries,
      ExecutorService executor, RetryPolicyConfig retryPolicyConfig) {

    RetryPolicy<Object> retryPolicy = new RetryPolicy<>()
        .withMaxRetries(retryPolicyConfig.getMaxRetries())
        .withBackoff(retryPolicyConfig.getBackoffDelay(),
            retryPolicyConfig.getBackoffMaxDelay(), retryPolicyConfig.getUnit(),
            retryPolicyConfig.getBackoffDelayFactor())
        .withMaxDuration(Duration.ofMillis(retryPolicyConfig.getMaxDuration()))
        .withJitter(retryPolicyConfig.getJitterFactor())
        .abortOn(abortRetries)
        .onRetry(e -> LOG.warn("Action: {} attempt: {} completed with error {} ms after start. Retrying up to {} ms.",
            opName, e.getAttemptCount(), e.getElapsedTime().toMillis(), retryPolicyConfig.getMaxDuration(),
            e.getLastFailure()));

    return Failsafe.with(retryPolicy).with(executor).getStageAsync(action::get);
  }
Helper method to convert: {@code Map<K, CompletableFuture<V>>}
to:                       {@code CompletableFuture<Map<K, V>>}

Returns a future that completes with successful map entries, skipping any entries with ignored errors,
when all value futures complete.
Returned future completes exceptionally if any of the futures complete with a non-ignored error.
samza-core/src/main/java/org/apache/samza/util/FutureUtil.java