Handle the Retry-After: header

johngmyers · johngmyers · commit 783164c89cd5 · 2025-02-22T11:12:09.000-08:00
diff --git a/CHANGES b/CHANGES
@@ -17,6 +17,8 @@ Platform 3.28
 
   We added 429 to the list of HTTP status codes that allow retrying requests.
 
+  We now handle the Retry-After: header of 429 HTTP status codes.
+
 Platform 3.27
 
 * Java version support
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/BackoffPolicy.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/BackoffPolicy.java
@@ -21,5 +21,5 @@ interface BackoffPolicy
 {
     BackoffPolicy nextAttempt();
 
-    Duration backoff(Duration previousBackoff);
+    Duration backoff(Duration previousBackoff, Duration suggestedBackoff);
 }
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/BalancingHttpClient.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/BalancingHttpClient.java
@@ -143,7 +143,7 @@ public <T, E extends Exception> T execute(Request request, ResponseHandler<T, E>
             }
             catch (RetryException e) {
                 attempt.markBad(e.getFailureCategory());
-                Duration backoff = attemptBackoffPolicy.backoff(previousBackoff);
+                Duration backoff = attemptBackoffPolicy.backoff(previousBackoff, e.getSuggestedBackoff());
                 long millis = backoff.roundTo(MILLISECONDS);
                 try {
                     Thread.sleep(millis);
@@ -316,7 +316,7 @@ else if (t instanceof RetryException retryException) {
                         attempt.markBad(retryException.getFailureCategory());
                         TraceToken traceToken = getCurrentTraceToken();
                         synchronized (subFutureLock) {
-                            Duration backoff = attemptBackoffPolicy.backoff(previousBackoff);
+                            Duration backoff = attemptBackoffPolicy.backoff(previousBackoff, retryException.getSuggestedBackoff());
                             ScheduledFuture<?> scheduledFuture = retryExecutor.schedule(() -> {
                                 try (TraceTokenScope scope = registerTraceToken(traceToken)){
                                     synchronized (subFutureLock) {
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/DecorrelatedJitteredBackoffPolicy.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/DecorrelatedJitteredBackoffPolicy.java
@@ -42,7 +42,7 @@ public BackoffPolicy nextAttempt()
     }
 
     @Override
-    public Duration backoff(Duration previousBackoff)
+    public Duration backoff(Duration previousBackoff, Duration suggestedBackoff)
     {
         long prev = previousBackoff.roundTo(TimeUnit.NANOSECONDS);
         long range = Math.abs(prev * 3 - min);
@@ -52,7 +52,7 @@ public Duration backoff(Duration previousBackoff)
         } else {
             randBackoff = min + ThreadLocalRandom.current().nextLong(range);
         }
-        long backoff = Math.min(max, randBackoff);
+        long backoff = Math.min(max, Math.max(randBackoff, suggestedBackoff.roundTo(TimeUnit.NANOSECONDS)));
 
         return new Duration(backoff, TimeUnit.NANOSECONDS);
     }
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/HttpServiceBalancerImpl.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/HttpServiceBalancerImpl.java
@@ -42,6 +42,7 @@
 import java.util.stream.Collectors;
 
 import static com.google.common.base.Preconditions.checkState;
+import static com.proofpoint.http.client.balancing.RetryException.NO_SUGGESTED_BACKOFF;
 import static java.util.Objects.requireNonNull;
 import static java.util.concurrent.TimeUnit.NANOSECONDS;
 import static java.util.concurrent.TimeUnit.SECONDS;
@@ -289,7 +290,7 @@ public void mark(boolean isFailure, InstanceState uriState, HttpServiceAttemptIm
                     if (++uriState.numFailures >= balancer.consecutiveFailures) {
                         uriState.liveness = DEAD;
                         uriState.backoffPolicy = balancer.backoffPolicy;
-                        uriState.lastBackoff = uriState.backoffPolicy.backoff(ZERO_DURATION);
+                        uriState.lastBackoff = uriState.backoffPolicy.backoff(ZERO_DURATION, NO_SUGGESTED_BACKOFF);
                         uriState.deadUntil = balancer.ticker.read() + uriState.lastBackoff.roundTo(NANOSECONDS);
                         balancer.httpServiceBalancerStats.removal(attempt.uri).add(uriState.lastBackoff);
                     }
@@ -321,7 +322,7 @@ public void mark(boolean isFailure, InstanceState uriState, HttpServiceAttemptIm
                 if (isFailure) {
                     uriState.liveness = DEAD;
                     uriState.backoffPolicy = uriState.backoffPolicy.nextAttempt();
-                    uriState.lastBackoff = uriState.backoffPolicy.backoff(uriState.lastBackoff);
+                    uriState.lastBackoff = uriState.backoffPolicy.backoff(uriState.lastBackoff, NO_SUGGESTED_BACKOFF);
                     uriState.deadUntil = balancer.ticker.read() + uriState.lastBackoff.roundTo(NANOSECONDS);
                     balancer.httpServiceBalancerStats.removal(attempt.uri).add(uriState.lastBackoff);
                 }
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/RetryException.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/RetryException.java
@@ -15,35 +15,52 @@
  */
 package com.proofpoint.http.client.balancing;
 
+import com.proofpoint.units.Duration;
+
+import java.util.concurrent.TimeUnit;
+
 class RetryException extends Exception
 {
+    static final Duration NO_SUGGESTED_BACKOFF = new Duration(0, TimeUnit.MILLISECONDS);
     private final String failureCategory;
+    private final Duration suggestedBackoff;
 
     RetryException(String failureCategory)
+    {
+        this(failureCategory, NO_SUGGESTED_BACKOFF);
+    }
+
+    RetryException(String failureCategory, Duration suggestedBackoff)
     {
         this.failureCategory = failureCategory;
+        this.suggestedBackoff = suggestedBackoff;
     }
 
     RetryException(Exception cause)
     {
         super(cause);
         failureCategory = cause.getClass().getSimpleName();
+        suggestedBackoff = NO_SUGGESTED_BACKOFF;
     }
 
     RetryException(Exception cause, String failureCategory)
     {
         super(cause);
         this.failureCategory = failureCategory;
+        suggestedBackoff = NO_SUGGESTED_BACKOFF;
     }
 
     RetryException(Exception cause, Exception failureException)
     {
         super(cause);
         failureCategory = failureException.getClass().getSimpleName();
+        suggestedBackoff = NO_SUGGESTED_BACKOFF;
     }
 
     String getFailureCategory()
     {
         return failureCategory;
     }
+
+    Duration getSuggestedBackoff() { return suggestedBackoff; }
 }
diff --git a/http-client/src/main/java/com/proofpoint/http/client/balancing/RetryingResponseHandler.java b/http-client/src/main/java/com/proofpoint/http/client/balancing/RetryingResponseHandler.java
@@ -22,11 +22,17 @@
 import com.proofpoint.http.client.Response;
 import com.proofpoint.http.client.ResponseHandler;
 import com.proofpoint.log.Logger;
+import com.proofpoint.units.Duration;
 
 import java.util.Set;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
 
+import static com.proofpoint.http.client.balancing.RetryException.NO_SUGGESTED_BACKOFF;
+import static java.lang.Integer.parseInt;
+import static java.util.concurrent.TimeUnit.SECONDS;
+
 final class RetryingResponseHandler<T, E extends Exception>
         implements ResponseHandler<T, RetryException>
 {
@@ -89,7 +95,19 @@ public T handle(Request request, Response response)
             log.warn("%d response querying %s",
                     response.getStatusCode(), request.getUri().resolve("/"));
             if (!("no".equalsIgnoreCase(retryHeader)) && bodySourceRetryable(request) && retryBudget.canRetry()) {
-                throw new RetryException(failureCategory);
+                Duration suggestedBackoff = NO_SUGGESTED_BACKOFF;
+                if (response.getStatusCode() == 429) {
+                    String retryAfterHeader = response.getHeader("Retry-After");
+                    if (retryAfterHeader != null) {
+                        try {
+                            suggestedBackoff = new Duration(parseInt(retryAfterHeader), SECONDS);
+                        }
+                        catch (NumberFormatException e) {
+                            // ignore
+                        }
+                    }
+                }
+                throw new RetryException(failureCategory, suggestedBackoff);
             }
 
             Object result;
diff --git a/http-client/src/test/java/com/proofpoint/http/client/balancing/TestDecorrelatedJitteredBackoffPolicy.java b/http-client/src/test/java/com/proofpoint/http/client/balancing/TestDecorrelatedJitteredBackoffPolicy.java
@@ -20,8 +20,10 @@
 
 import java.util.concurrent.TimeUnit;
 
+import static com.proofpoint.http.client.balancing.RetryException.NO_SUGGESTED_BACKOFF;
 import static com.proofpoint.testing.Assertions.assertGreaterThanOrEqual;
 import static com.proofpoint.testing.Assertions.assertLessThanOrEqual;
+import static org.testng.Assert.assertEquals;
 
 public class TestDecorrelatedJitteredBackoffPolicy
 {
@@ -35,7 +37,7 @@ public void testBackoff()
             Duration last = new Duration(0, TimeUnit.SECONDS);
             Duration prev = min; // Special-case constraint for the first backoff
             for (int j = 0; j < 1000; j++) {
-                Duration next = policy.backoff(last);
+                Duration next = policy.backoff(last, NO_SUGGESTED_BACKOFF);
                 assertGreaterThanOrEqual(next, min);
                 assertLessThanOrEqual(next, new Duration(prev.roundTo(TimeUnit.NANOSECONDS) * 3, TimeUnit.NANOSECONDS));
                 assertLessThanOrEqual(next, max);
@@ -44,4 +46,39 @@ public void testBackoff()
             }
         }
     }
+
+    @Test
+    public void testSuggestedBackoff()
+    {
+        Duration min = new Duration(1, TimeUnit.MILLISECONDS);
+        Duration suggested = new Duration(1, TimeUnit.SECONDS);
+        Duration max = new Duration(5, TimeUnit.SECONDS);
+        BackoffPolicy policy = new DecorrelatedJitteredBackoffPolicy(min, max);
+        for (int i = 0; i < 1000; i++) {
+            Duration last = new Duration(0, TimeUnit.SECONDS);
+            Duration prev = min; // Special-case constraint for the first backoff
+            for (int j = 0; j < 1000; j++) {
+                Duration next = policy.backoff(last, suggested);
+                assertGreaterThanOrEqual(next, suggested);
+                if (next.getValue(TimeUnit.NANOSECONDS) > suggested.getValue(TimeUnit.NANOSECONDS)) {
+                    assertLessThanOrEqual(next, new Duration(prev.roundTo(TimeUnit.NANOSECONDS) * 3, TimeUnit.NANOSECONDS));
+                }
+                assertLessThanOrEqual(next, max);
+                prev = next;
+                last = next;
+            }
+        }
+    }
+
+    @Test
+    public void testSuggestedBackoffGreaterThanMax()
+    {
+        Duration min = new Duration(1, TimeUnit.MILLISECONDS);
+        Duration suggested = new Duration(10, TimeUnit.SECONDS);
+        Duration max = new Duration(5, TimeUnit.SECONDS);
+        BackoffPolicy policy = new DecorrelatedJitteredBackoffPolicy(min, max);
+        Duration last = new Duration(0, TimeUnit.SECONDS);
+        Duration next = policy.backoff(last, suggested);
+        assertEquals(next, max);
+    }
 }

Original file line number	Diff line number	Diff line change
`@@ -21,5 +21,5 @@ interface BackoffPolicy`
`21`	`21`	`{`
`22`	`22`	`BackoffPolicy nextAttempt();`
`23`	`23`
`24`		`- Duration backoff(Duration previousBackoff);`
	`24`	`+ Duration backoff(Duration previousBackoff, Duration suggestedBackoff);`
`25`	`25`	`}`
Original file line number	Diff line number	Diff line change
`@@ -42,7 +42,7 @@ public BackoffPolicy nextAttempt()`
`42`	`42`	`}`
`43`	`43`
`44`	`44`	`@Override`
`45`		`- public Duration backoff(Duration previousBackoff)`
	`45`	`+ public Duration backoff(Duration previousBackoff, Duration suggestedBackoff)`
`46`	`46`	`{`
`47`	`47`	`long prev = previousBackoff.roundTo(TimeUnit.NANOSECONDS);`
`48`	`48`	`long range = Math.abs(prev * 3 - min);`
`@@ -52,7 +52,7 @@ public Duration backoff(Duration previousBackoff)`
`52`	`52`	`} else {`
`53`	`53`	`randBackoff = min + ThreadLocalRandom.current().nextLong(range);`
`54`	`54`	`}`
`55`		`- long backoff = Math.min(max, randBackoff);`
	`55`	`+ long backoff = Math.min(max, Math.max(randBackoff, suggestedBackoff.roundTo(TimeUnit.NANOSECONDS)));`
`56`	`56`
`57`	`57`	`return new Duration(backoff, TimeUnit.NANOSECONDS);`
`58`	`58`	`}`
Original file line number	Diff line number	Diff line change
`@@ -15,35 +15,52 @@`
`15`	`15`	`*/`
`16`	`16`	`package com.proofpoint.http.client.balancing;`
`17`	`17`
	`18`	`+import com.proofpoint.units.Duration;`
	`19`	`+`
	`20`	`+import java.util.concurrent.TimeUnit;`
	`21`	`+`
`18`	`22`	`class RetryException extends Exception`
`19`	`23`	`{`
	`24`	`+ static final Duration NO_SUGGESTED_BACKOFF = new Duration(0, TimeUnit.MILLISECONDS);`
`20`	`25`	`private final String failureCategory;`
	`26`	`+ private final Duration suggestedBackoff;`
`21`	`27`
`22`	`28`	`RetryException(String failureCategory)`
	`29`	`+ {`
	`30`	`+ this(failureCategory, NO_SUGGESTED_BACKOFF);`
	`31`	`+ }`
	`32`	`+`
	`33`	`+ RetryException(String failureCategory, Duration suggestedBackoff)`
`23`	`34`	`{`
`24`	`35`	`this.failureCategory = failureCategory;`
	`36`	`+ this.suggestedBackoff = suggestedBackoff;`
`25`	`37`	`}`
`26`	`38`
`27`	`39`	`RetryException(Exception cause)`
`28`	`40`	`{`
`29`	`41`	`super(cause);`
`30`	`42`	`failureCategory = cause.getClass().getSimpleName();`
	`43`	`+ suggestedBackoff = NO_SUGGESTED_BACKOFF;`
`31`	`44`	`}`
`32`	`45`
`33`	`46`	`RetryException(Exception cause, String failureCategory)`
`34`	`47`	`{`
`35`	`48`	`super(cause);`
`36`	`49`	`this.failureCategory = failureCategory;`
	`50`	`+ suggestedBackoff = NO_SUGGESTED_BACKOFF;`
`37`	`51`	`}`
`38`	`52`
`39`	`53`	`RetryException(Exception cause, Exception failureException)`
`40`	`54`	`{`
`41`	`55`	`super(cause);`
`42`	`56`	`failureCategory = failureException.getClass().getSimpleName();`
	`57`	`+ suggestedBackoff = NO_SUGGESTED_BACKOFF;`
`43`	`58`	`}`
`44`	`59`
`45`	`60`	`String getFailureCategory()`
`46`	`61`	`{`
`47`	`62`	`return failureCategory;`
`48`	`63`	`}`
	`64`	`+`
	`65`	`+ Duration getSuggestedBackoff() { return suggestedBackoff; }`
`49`	`66`	`}`