Introduce RetryLimiter (line#6409)

Motivation:

This changeset attempts to solve the same problem as line#6318.

Retry limiting is a concept which limits the number of retries in case a
system undergoes a prolonged period of service degradation.
gRPC offers a
[token-based](https://github.com/grpc/proposal/blob/master/A6-client-retries.md#throttling-retry-attempts-and-hedged-rpcs)
configuration which limits retries depending on certain predicates,
whereas envoy offers a simple [concurrency limiting
configuration](https://www.envoyproxy.io/docs/envoy/latest/api-v3/config/cluster/v3/circuit_breaker.proto#envoy-v3-api-msg-config-cluster-v3-circuitbreakers-thresholds-retrybudget)
based on the number of active retries.

To support token-based retry limiters, I propose that
`RetryDecision#permits` is added as metadata which could signal to
`RetryLimiter` whether retries should be further made. This could be
useful for systems behind load balancers, as the load balancer may
return certain status codes depending on the health upstream.

To support simple concurrency-based retry limiters, I propose that a
`RetryLimiter#shouldRetry` method is called right before a retry is
executed.

Modifications:

- Introduced `RetryLimiter` which acts as an extension to dynamically
limit retries.
- `RetryLimiter#shouldRetry` decides whether a retry should be executed
- `RetryLimiter#handleDecision` is invoked when a `RetryDecision` is
made. `RetryDecision#permits` may be used to update the internal state
and decide whether retries should be allowed.
- Added `RetryDecision#permits`
- Added APIs so users can set `RetryLimiter` to `RetryConfig`

Result:

- Users can specify `RetryLimiter` to limit retry requests.
- Closes line#6282

<!--
Visit this URL to learn more about how to write a pull request
description:

https://armeria.dev/community/developer-guide#how-to-write-pull-request-description
-->

Author: jrhee17

core/src/main/java/com/linecorp/armeria/client/retry/ConcurrencyBasedRetryLimiter.java

@@ -0,0 +1,55 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.client.retry;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import java.util.concurrent.atomic.AtomicLong;
+
+import com.google.common.base.MoreObjects;
+
+import com.linecorp.armeria.client.ClientRequestContext;
+
+final class ConcurrencyBasedRetryLimiter implements RetryLimiter {
+
+    private final long maxRequests;
+    private final AtomicLong activeRequests = new AtomicLong();
+
+    ConcurrencyBasedRetryLimiter(long maxRequests) {
+        checkArgument(maxRequests > 0, "maxRequests must be positive: %s.", maxRequests);
+        this.maxRequests = maxRequests;
+    }
+
+    @Override
+    public boolean shouldRetry(ClientRequestContext ctx) {
+        final long cnt = activeRequests.incrementAndGet();
+        if (cnt > maxRequests) {
+            activeRequests.decrementAndGet();
+            return false;
+        }
+        ctx.log().whenComplete().thenRun(activeRequests::decrementAndGet);
+        return true;
+    }
+
+    @Override
+    public String toString() {
+        return MoreObjects.toStringHelper(this)
+                          .add("activeRequests", activeRequests)
+                          .add("maxRequests", maxRequests)
+                          .toString();
+    }
+}

core/src/main/java/com/linecorp/armeria/client/retry/RetryConfig.java

@@ -78,6 +78,7 @@ static <T extends Response> RetryConfigBuilder<T> builder0(
     private final int maxTotalAttempts;
     private final long responseTimeoutMillisForEachAttempt;
     private final int maxContentLength;
+    private final RetryLimiter retryLimiter;
 
     @Nullable
     private final RetryRule retryRule;
@@ -88,27 +89,30 @@ static <T extends Response> RetryConfigBuilder<T> builder0(
     @Nullable
     private RetryRuleWithContent<T> fromRetryRule;
 
-    RetryConfig(RetryRule retryRule, int maxTotalAttempts, long responseTimeoutMillisForEachAttempt) {
+    RetryConfig(RetryRule retryRule, int maxTotalAttempts, long responseTimeoutMillisForEachAttempt,
+                RetryLimiter retryLimiter) {
         this(requireNonNull(retryRule, "retryRule"), null,
-                maxTotalAttempts, responseTimeoutMillisForEachAttempt, 0);
+                maxTotalAttempts, responseTimeoutMillisForEachAttempt, 0, retryLimiter);
         checkArguments(maxTotalAttempts, responseTimeoutMillisForEachAttempt);
     }
 
     RetryConfig(
             RetryRuleWithContent<T> retryRuleWithContent,
             int maxContentLength,
             int maxTotalAttempts,
-            long responseTimeoutMillisForEachAttempt) {
+            long responseTimeoutMillisForEachAttempt,
+            RetryLimiter retryLimiter) {
         this(null, requireNonNull(retryRuleWithContent, "retryRuleWithContent"),
-                maxTotalAttempts, responseTimeoutMillisForEachAttempt, maxContentLength);
+                maxTotalAttempts, responseTimeoutMillisForEachAttempt, maxContentLength, retryLimiter);
     }
 
     private RetryConfig(
             @Nullable RetryRule retryRule,
             @Nullable RetryRuleWithContent<T> retryRuleWithContent,
             int maxTotalAttempts,
             long responseTimeoutMillisForEachAttempt,
-            int maxContentLength) {
+            int maxContentLength, RetryLimiter retryLimiter) {
+        this.retryLimiter = new RetryLimiters.CatchingRetryLimiter(retryLimiter);
         checkArguments(maxTotalAttempts, responseTimeoutMillisForEachAttempt);
         this.retryRule = retryRule;
         this.retryRuleWithContent = retryRuleWithContent;
@@ -147,7 +151,8 @@ public RetryConfigBuilder<T> toBuilder() {
         }
         return builder
                 .maxTotalAttempts(maxTotalAttempts)
-                .responseTimeoutMillisForEachAttempt(responseTimeoutMillisForEachAttempt);
+                .responseTimeoutMillisForEachAttempt(responseTimeoutMillisForEachAttempt)
+                .retryLimiter(retryLimiter);
     }
 
     /**
@@ -197,6 +202,13 @@ public boolean needsContentInRule() {
         return retryRuleWithContent != null;
     }
 
+    /**
+     * Returns the configured {@link RetryLimiter}.
+     */
+    public RetryLimiter retryLimiter() {
+        return retryLimiter;
+    }
+
     /**
      * Returns whether the associated {@link RetryRule} or {@link RetryRuleWithContent} requires
      * response trailers.

core/src/main/java/com/linecorp/armeria/client/retry/RetryConfigBuilder.java

@@ -24,6 +24,7 @@
 import com.google.common.base.MoreObjects;
 import com.google.common.base.MoreObjects.ToStringHelper;
 
+import com.linecorp.armeria.client.retry.RetryLimiters.AlwaysRetryLimiter;
 import com.linecorp.armeria.common.Flags;
 import com.linecorp.armeria.common.Response;
 import com.linecorp.armeria.common.annotation.Nullable;
@@ -42,6 +43,7 @@ public final class RetryConfigBuilder<T extends Response> {
     private final RetryRule retryRule;
     @Nullable
     private final RetryRuleWithContent<T> retryRuleWithContent;
+    private RetryLimiter retryLimiter = AlwaysRetryLimiter.INSTANCE;
 
     /**
      * Creates a {@link RetryConfigBuilder} with this {@link RetryRule}.
@@ -111,19 +113,30 @@ public RetryConfigBuilder<T> responseTimeoutForEachAttempt(Duration responseTime
         return this;
     }
 
+    /**
+     * Sets a {@link RetryLimiter} which may limit retry requests.
+     * @see RetryLimiter
+     */
+    public RetryConfigBuilder<T> retryLimiter(RetryLimiter retryLimiter) {
+        this.retryLimiter = requireNonNull(retryLimiter, "retryLimiter");
+        return this;
+    }
+
     /**
      * Returns a newly-created {@link RetryConfig} from this {@link RetryConfigBuilder}'s values.
      */
     public RetryConfig<T> build() {
         if (retryRule != null) {
-            return new RetryConfig<>(retryRule, maxTotalAttempts, responseTimeoutMillisForEachAttempt);
+            return new RetryConfig<>(retryRule, maxTotalAttempts, responseTimeoutMillisForEachAttempt,
+                                     retryLimiter);
         }
         assert retryRuleWithContent != null;
         return new RetryConfig<>(
                 retryRuleWithContent,
                 maxContentLength,
                 maxTotalAttempts,
-                responseTimeoutMillisForEachAttempt);
+                responseTimeoutMillisForEachAttempt,
+                retryLimiter);
     }
 
     @Override
@@ -139,6 +152,7 @@ ToStringHelper toStringHelper() {
                 .add("retryRuleWithContent", retryRuleWithContent)
                 .add("maxTotalAttempts", maxTotalAttempts)
                 .add("responseTimeoutMillisForEachAttempt", responseTimeoutMillisForEachAttempt)
-                .add("maxContentLength", maxContentLength);
+                .add("maxContentLength", maxContentLength)
+                .add("retryLimiter", retryLimiter);
     }
 }

core/src/main/java/com/linecorp/armeria/client/retry/RetryDecision.java

@@ -18,6 +18,9 @@
 
 import static java.util.Objects.requireNonNull;
 
+import com.google.common.base.MoreObjects;
+import com.google.common.base.MoreObjects.ToStringHelper;
+
 import com.linecorp.armeria.common.annotation.Nullable;
 
 /**
@@ -26,27 +29,44 @@
  */
 public final class RetryDecision {
 
-    private static final RetryDecision NO_RETRY = new RetryDecision(null);
-    private static final RetryDecision NEXT = new RetryDecision(null);
-    static final RetryDecision DEFAULT = new RetryDecision(Backoff.ofDefault());
+    private static final RetryDecision NO_RETRY = new RetryDecision(null, -1);
+    private static final RetryDecision NEXT = new RetryDecision(null, 0);
+    static final RetryDecision DEFAULT = new RetryDecision(Backoff.ofDefault(), 1);
 
     /**
      * Returns a {@link RetryDecision} that retries with the specified {@link Backoff}.
+     * The permits will be {@code 1} by default.
      */
     public static RetryDecision retry(Backoff backoff) {
-        if (backoff == Backoff.ofDefault()) {
+        return retry(backoff, 1);
+    }
+
+    /**
+     * Returns a {@link RetryDecision} that retries with the specified {@link Backoff}.
+     */
+    @SuppressWarnings("FloatingPointEquality")
+    public static RetryDecision retry(Backoff backoff, double permits) {
+        if (backoff == DEFAULT.backoff() && permits == DEFAULT.permits()) {
             return DEFAULT;
         }
-        return new RetryDecision(requireNonNull(backoff, "backoff"));
+        return new RetryDecision(requireNonNull(backoff, "backoff"), permits);
     }
 
     /**
      * Returns a {@link RetryDecision} that never retries.
+     * The permits will be {@code -1} by default.
      */
     public static RetryDecision noRetry() {
         return NO_RETRY;
     }
 
+    /**
+     * Returns a {@link RetryDecision} that never retries.
+     */
+    public static RetryDecision noRetry(double permits) {
+        return new RetryDecision(null, permits);
+    }
+
     /**
      * Returns a {@link RetryDecision} that skips the current {@link RetryRule} and
      * tries to retry with the next {@link RetryRule}.
@@ -57,24 +77,41 @@ public static RetryDecision next() {
 
     @Nullable
     private final Backoff backoff;
+    private final double permits;
 
-    private RetryDecision(@Nullable Backoff backoff) {
+    private RetryDecision(@Nullable Backoff backoff, double permits) {
         this.backoff = backoff;
+        this.permits = permits;
     }
 
     @Nullable
     Backoff backoff() {
         return backoff;
     }
 
+    /**
+     * The number of permits associated with this {@link RetryDecision}.
+     * This may be used by {@link RetryLimiter} to determine whether retry requests should
+     * be limited or not. The semantics of whether or how the returned value affects {@link RetryLimiter}
+     * depends on what type of {@link RetryLimiter} is used.
+     */
+    public double permits() {
+        return permits;
+    }
+
     @Override
     public String toString() {
-        if (this == NO_RETRY) {
-            return "RetryDecision(NO_RETRY)";
-        } else if (this == NEXT) {
-            return "RetryDecision(NEXT)";
+        final ToStringHelper stringHelper = MoreObjects.toStringHelper(this);
+        if (this == NEXT) {
+            stringHelper.add("type", "NEXT");
+        } else if (backoff != null) {
+            stringHelper.add("type", "RETRY");
         } else {
-            return "RetryDecision(RETRY(" + backoff + "))";
+            stringHelper.add("type", "NO_RETRY");
         }
+        return stringHelper.omitNullValues()
+                           .add("backoff", backoff)
+                           .add("permits", permits)
+                           .toString();
     }
 }

core/src/main/java/com/linecorp/armeria/client/retry/RetryLimitedException.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2025 LY Corporation
+ *
+ * LY Corporation licenses this file to you under the Apache License,
+ * version 2.0 (the "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at:
+ *
+ *   https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+ * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+ * License for the specific language governing permissions and limitations
+ * under the License.
+ */
+
+package com.linecorp.armeria.client.retry;
+
+import com.linecorp.armeria.common.Flags;
+
+/**
+ * An exception thrown when a retry is limited by a {@link RetryLimiter}.
+ */
+public final class RetryLimitedException extends RuntimeException {
+
+    private static final long serialVersionUID = 7203512016805562689L;
+
+    private static final RetryLimitedException INSTANCE = new RetryLimitedException(false);
+
+    /**
+     * Returns an instance of {@link RetryLimitedException} sampled by {@link Flags#verboseExceptionSampler()}.
+     */
+    public static RetryLimitedException of() {
+        return isSampled() ? new RetryLimitedException(true) : INSTANCE;
+    }
+
+    private RetryLimitedException(boolean enableSuppression) {
+        super(null, null, enableSuppression, isSampled());
+    }
+
+    private static boolean isSampled() {
+        return Flags.verboseExceptionSampler().isSampled(RetryLimitedException.class);
+    }
+}