static <T> FailsafeExecutor<T> failsafe(RetryPolicy<T> retryPolicy, RetryMetrics metrics, ScheduledExecutorService retryExec) {
    long startMs = System.currentTimeMillis();

    RetryPolicy<T> retryPolicyWithMetrics = retryPolicy
        .onRetry(e -> metrics.retryCount.inc())
        .onRetriesExceeded(e -> {
          metrics.retryTimer.update(System.currentTimeMillis() - startMs);
          metrics.permFailureCount.inc();
        }).onSuccess((e) -> {
          if (e.getAttemptCount() > 1) {
            metrics.retryTimer.update(System.currentTimeMillis() - startMs);
          } else {
            metrics.successCount.inc();
          }
        });

    return Failsafe.with(retryPolicyWithMetrics).with(retryExec);
  }
Obtain an async failsafe retryer instance with the specified policy, metrics, and executor service.
@param retryPolicy retry policy
@param metrics retry metrics
@param retryExec executor service for scheduling async retries
@return {@link net.jodah.failsafe.FailsafeExecutor} instance
samza-core/src/main/java/org/apache/samza/table/retry/FailsafeAdapter.java