Enable retry throttling

[schiemon]

This issue originates from a user's question on Discord.

gRPC supports retry throttling via a token-based mechanism: https://github.com/grpc/proposal/blob/master/A6-client-retries.md#throttling-retry-attempts-and-hedged-rpcs. Essentially it prevents excessive retries on endpoints that proved to fail requests in the past.

This issue is opened to close the gap with gRPC by adding retry throttling support in Retrying(Rpc)Client and/or RetryRule.

@ikhoon proposed to add following API to RetryConfig:

public interface RetryLimiter {

    static RetryLimiter ofRateLimiter(double permitsPerSecond) {
        return new DefaultRetryLimiter(permitsPerSecond);
    }

    /**
     * Determines whether the request should be retried.
     * This method is not invoked:
     * - if the maximum number of attempts has been reached.
     * - if the request has been cancelled.
     * - if the request is the first attempt.
     */
    boolean shouldRetry(ClientRequestContext ctx, int numAttemptsSoFar);

    /**
     * Invoked when an attempt is successful.
     * @param numAttemptsSoFar the number of attempts made so far, including the current one
     */
    void onSuccess(ClientRequestContext ctx, int numAttemptsSoFar);

    /**
     * Invoked when an attempt fails.
     *
     * @param numAttemptsSoFar the number of attempts made so far, including the current one
     */
    void onFailure(ClientRequestContext ctx, int numAttemptsSoFar);
}

final class DefaultRetryLimiter implements RetryLimiter {
    // Use Guava's RateLimiter as the default limiter
    private final RateLimiter rateLimiter;

    DefaultRetryLimiter(double permitsPerSecond) {
        rateLimiter = RateLimiter.create(permitsPerSecond);
    }

    @Override
    public boolean shouldRetry(ClientRequestContext ctx, int numAttemptsSoFar) {
        return rateLimiter.tryAcquire();
    }

    @Override
    public void onSuccess(ClientRequestContext ctx, int numAttemptsSoFar) {}

    @Override
    public void onFailure(ClientRequestContext ctx, int numAttemptsSoFar) {}
}

public final class RetryConfigBuilder<T extends Response> {
    public RetryConfigBuilder<T> retryLimiter(long permitsPerSecond) {
        retryLimiter = RetryLimiter.ofRateLimiter(permitsPerSecond);
        return this;
    }

    public RetryConfigBuilder<T> rateLimiter(RetryLimiter retryLimiter) {
        this.retryLimiter = retryLimiter;
        return this;
    }
}

which can be added to RetryConfig like follows:

RetryConfig
  .builder(retryRule)
  .retryLimiter(1000)
  ...
  .build()