Respect Retry-After in OTLP HTTP senders

exporters/common/src/main/java/io/opentelemetry/exporter/internal/RetryUtil.java

@@ -6,11 +6,19 @@
 package io.opentelemetry.exporter.internal;
 
 import io.opentelemetry.sdk.common.export.GrpcStatusCode;
+import java.time.Duration;
+import java.time.Instant;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeParseException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.OptionalLong;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
 import java.util.stream.Collectors;
+import javax.annotation.Nullable;
 
 /**
  * This class is internal and is hence not for public use. Its APIs are unstable and can change at
@@ -47,4 +55,42 @@ public static Set<String> retryableGrpcStatusCodes() {
   public static Set<Integer> retryableHttpResponseCodes() {
     return RETRYABLE_HTTP_STATUS_CODES;
   }
+
+  /** Returns whether the status code represents OTLP/HTTP throttling. */
+  public static boolean isOtlpHttpThrottleStatusCode(int statusCode) {
+    return statusCode == 429 || statusCode == 503;
+  }
+
+  /**
+   * Returns the delay specified by a {@code Retry-After} header, or empty if the value is absent or
+   * malformed.
+   */
+  public static OptionalLong retryAfterNanos(@Nullable String retryAfter) {
+    return retryAfterNanos(retryAfter, Instant.now());
+  }
+
+  static OptionalLong retryAfterNanos(@Nullable String retryAfter, Instant now) {
+    if (retryAfter == null) {
+      return OptionalLong.empty();
+    }
+
+    try {
+      long delaySeconds = Long.parseLong(retryAfter);
+      if (delaySeconds < 0) {
+        return OptionalLong.empty();
+      }
+      return OptionalLong.of(TimeUnit.SECONDS.toNanos(delaySeconds));
+    } catch (NumberFormatException ignored) {
+      // Fall through and try the HTTP-date form.
+    }
+
+    try {
+      Instant retryAfterInstant =
+          ZonedDateTime.parse(retryAfter, DateTimeFormatter.RFC_1123_DATE_TIME).toInstant();
+      long delayNanos = Duration.between(now, retryAfterInstant).toNanos();
+      return OptionalLong.of(Math.max(0, delayNanos));
+    } catch (DateTimeParseException | ArithmeticException ignored) {
+      return OptionalLong.empty();
+    }
+  }
 }

exporters/common/src/test/java/io/opentelemetry/exporter/internal/RetryUtilTest.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.exporter.internal;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+import java.time.Instant;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
+import java.util.OptionalLong;
+import java.util.concurrent.TimeUnit;
+import org.junit.jupiter.api.Test;
+
+class RetryUtilTest {
+
+  @Test
+  void retryAfterSeconds() {
+    OptionalLong delayNanos = RetryUtil.retryAfterNanos("30", Instant.EPOCH);
+
+    assertThat(delayNanos).hasValue(TimeUnit.SECONDS.toNanos(30));
+  }
+
+  @Test
+  void retryAfterDate() {
+    Instant now = Instant.parse("2026-07-17T00:00:00Z");
+    String retryAfter =
+        ZonedDateTime.ofInstant(now.plusSeconds(45), ZoneOffset.UTC)
+            .format(DateTimeFormatter.RFC_1123_DATE_TIME);
+
+    OptionalLong delayNanos = RetryUtil.retryAfterNanos(retryAfter, now);
+
+    assertThat(delayNanos).hasValue(TimeUnit.SECONDS.toNanos(45));
+  }
+
+  @Test
+  void retryAfterPastDateClampsToZero() {
+    Instant now = Instant.parse("2026-07-17T00:00:00Z");
+    String retryAfter =
+        ZonedDateTime.ofInstant(now.minusSeconds(1), ZoneOffset.UTC)
+            .format(DateTimeFormatter.RFC_1123_DATE_TIME);
+
+    OptionalLong delayNanos = RetryUtil.retryAfterNanos(retryAfter, now);
+
+    assertThat(delayNanos).hasValue(0L);
+  }
+
+  @Test
+  void retryAfterMalformed() {
+    OptionalLong delayNanos = RetryUtil.retryAfterNanos("bad-value", Instant.EPOCH);
+
+    assertThat(delayNanos).isEmpty();
+  }
+}

exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/AbstractHttpTelemetryExporterTest.java

@@ -58,6 +58,9 @@
 import java.nio.file.Files;
 import java.security.cert.CertificateEncodingException;
 import java.time.Duration;
+import java.time.ZoneOffset;
+import java.time.ZonedDateTime;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Base64;
@@ -729,6 +732,52 @@ void retryableError_tooManyAttempts() {
     assertThat(attempts).hasValue(2);
   }
 
+  @Test
+  void retryableThrottledError_retryAfterSeconds() {
+    addHttpResponse(429, "0");
+
+    assertThat(
+            exporter
+                .export(Collections.singletonList(generateFakeTelemetry()))
+                .join(10, TimeUnit.SECONDS)
+                .isSuccess())
+        .isTrue();
+
+    assertThat(attempts).hasValue(2);
+  }
+
+  @Test
+  void retryableThrottledError_retryAfterDate() {
+    addHttpResponse(
+        503,
+        ZonedDateTime.now(ZoneOffset.UTC)
+            .minusSeconds(1)
+            .format(DateTimeFormatter.RFC_1123_DATE_TIME));
+
+    assertThat(
+            exporter
+                .export(Collections.singletonList(generateFakeTelemetry()))
+                .join(10, TimeUnit.SECONDS)
+                .isSuccess())
+        .isTrue();
+
+    assertThat(attempts).hasValue(2);
+  }
+
+  @Test
+  void retryableThrottledError_malformedRetryAfterFallsBack() {
+    addHttpResponse(503, "not-a-retry-after");
+
+    assertThat(
+            exporter
+                .export(Collections.singletonList(generateFakeTelemetry()))
+                .join(10, TimeUnit.SECONDS)
+                .isSuccess())
+        .isTrue();
+
+    assertThat(attempts).hasValue(2);
+  }
+
   @ParameterizedTest
   @SuppressLogger(HttpExporter.class)
   @ValueSource(ints = {400, 401, 403, 500, 501})
@@ -1208,6 +1257,14 @@ private static void addHttpResponse(int code) {
     httpErrors.add(HttpResponse.of(code));
   }
 
+  private static void addHttpResponse(int code, String retryAfter) {
+    httpErrors.add(
+        HttpResponse.of(
+            ResponseHeaders.builder(HttpStatus.valueOf(code))
+                .add("Retry-After", retryAfter)
+                .build()));
+  }
+
   private static void addHttpResponse(int code, AbstractMessageLite<?, ?> bodyMessage) {
     httpErrors.add(
         HttpResponse.of(

exporters/sender/jdk/src/main/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSender.java

@@ -5,6 +5,7 @@
 
 package io.opentelemetry.exporter.sender.jdk.internal;
 
+import io.opentelemetry.exporter.internal.RetryUtil;
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.Compressor;
 import io.opentelemetry.sdk.common.export.HttpResponse;
@@ -27,6 +28,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.OptionalLong;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
@@ -54,7 +56,7 @@
  */
 public final class JdkHttpSender implements HttpSender {
 
-  private static final Set<Integer> retryableStatusCodes = Set.of(429, 502, 503, 504);
+  private static final Set<Integer> retryableStatusCodes = RetryUtil.retryableHttpResponseCodes();
 
   private static final ThreadLocal<NoCopyByteArrayOutputStream> threadLocalBaos =
       ThreadLocal.withInitial(NoCopyByteArrayOutputStream::new);
@@ -74,6 +76,8 @@ public final class JdkHttpSender implements HttpSender {
   @Nullable private final RetryPolicy retryPolicy;
   private final Predicate<IOException> retryExceptionPredicate;
   private final long maxResponseBodySize;
+  private final Sleeper sleeper;
+  private final Supplier<Double> randomJitter;
 
   // Visible for testing
   JdkHttpSender(
@@ -86,6 +90,33 @@ public final class JdkHttpSender implements HttpSender {
       @Nullable RetryPolicy retryPolicy,
       @Nullable ExecutorService executorService,
       long maxResponseBodySize) {
+    this(
+        client,
+        endpoint,
+        contentType,
+        compressor,
+        timeout,
+        headerSupplier,
+        retryPolicy,
+        executorService,
+        maxResponseBodySize,
+        TimeUnit.NANOSECONDS::sleep,
+        () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d));
+  }
+
+  // Visible for testing
+  JdkHttpSender(
+      HttpClient client,
+      URI endpoint,
+      String contentType,
+      @Nullable Compressor compressor,
+      Duration timeout,
+      Supplier<Map<String, List<String>>> headerSupplier,
+      @Nullable RetryPolicy retryPolicy,
+      @Nullable ExecutorService executorService,
+      long maxResponseBodySize,
+      Sleeper sleeper,
+      Supplier<Double> randomJitter) {
     this.client = client;
     this.endpoint = endpoint;
     this.contentType = contentType;
@@ -105,6 +136,8 @@ public final class JdkHttpSender implements HttpSender {
       this.managedExecutor = false;
     }
     this.maxResponseBodySize = maxResponseBodySize;
+    this.sleeper = sleeper;
+    this.randomJitter = randomJitter;
   }
 
   JdkHttpSender(
@@ -212,23 +245,27 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {
 
     // If no retry policy, short circuit
     if (retryPolicy == null) {
-      return sendRequest(requestBuilder, byteBufferPool);
+      return toHttpResponse(sendRequest(requestBuilder, byteBufferPool));
     }
 
     long attempt = 0;
     long nextBackoffNanos = retryPolicy.getInitialBackoff().toNanos();
     HttpResponse httpResponse = null;
     IOException exception = null;
+    OptionalLong retryDelayNanos = OptionalLong.empty();
     do {
       if (attempt > 0) {
         // Compute and sleep for backoff
         long currentBackoffNanos =
             Math.min(nextBackoffNanos, retryPolicy.getMaxBackoff().toNanos());
         long backoffNanos =
-            (long) (ThreadLocalRandom.current().nextDouble(0.8d, 1.2d) * currentBackoffNanos);
+            retryDelayNanos.isPresent()
+                ? retryDelayNanos.getAsLong()
+                : (long) (randomJitter.get() * currentBackoffNanos);
         nextBackoffNanos = (long) (currentBackoffNanos * retryPolicy.getBackoffMultiplier());
+        retryDelayNanos = OptionalLong.empty();
         try {
-          TimeUnit.NANOSECONDS.sleep(backoffNanos);
+          sleeper.sleep(backoffNanos);
         } catch (InterruptedException e) {
           Thread.currentThread().interrupt();
           break; // Break out and return response or throw
@@ -243,8 +280,10 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {
       exception = null;
       requestBuilder.timeout(timeout.minusNanos(System.nanoTime() - startTimeNanos));
       try {
-        httpResponse = sendRequest(requestBuilder, byteBufferPool);
-        boolean retryable = retryableStatusCodes.contains(httpResponse.getStatusCode());
+        java.net.http.HttpResponse<InputStream> rawResponse =
+            sendRequest(requestBuilder, byteBufferPool);
+        httpResponse = toHttpResponse(rawResponse);
+        boolean retryable = retryableStatusCodes.contains(rawResponse.statusCode());
         if (logger.isLoggable(Level.FINER)) {
           logger.log(
               Level.FINER,
@@ -258,6 +297,7 @@ HttpResponse sendInternal(MessageWriter requestBodyWriter) throws IOException {
         if (!retryable) {
           return httpResponse;
         }
+        retryDelayNanos = retryDelayNanos(rawResponse);
       } catch (IOException e) {
         exception = e;
         boolean retryable = retryExceptionPredicate.test(exception);
@@ -287,12 +327,10 @@ private static String responseStringRepresentation(HttpResponse response) {
     return "HttpResponse{code=" + response.getStatusCode() + "}";
   }
 
-  private HttpResponse sendRequest(
+  private java.net.http.HttpResponse<InputStream> sendRequest(
       HttpRequest.Builder requestBuilder, ByteBufferPool byteBufferPool) throws IOException {
     try {
-      java.net.http.HttpResponse<InputStream> response =
-          client.send(requestBuilder.build(), BodyHandlers.ofInputStream());
-      return toHttpResponse(response);
+      return client.send(requestBuilder.build(), BodyHandlers.ofInputStream());
     } catch (InterruptedException e) {
       Thread.currentThread().interrupt();
       throw new IllegalStateException(e);
@@ -412,6 +450,17 @@ private void resetPool() {
     }
   }
 
+  private static OptionalLong retryDelayNanos(java.net.http.HttpResponse<?> response) {
+    if (!RetryUtil.isOtlpHttpThrottleStatusCode(response.statusCode())) {
+      return OptionalLong.empty();
+    }
+    return RetryUtil.retryAfterNanos(response.headers().firstValue("Retry-After").orElse(null));
+  }
+
+  interface Sleeper {
+    void sleep(long delayNanos) throws InterruptedException;
+  }
+
   @Override
   public CompletableResultCode shutdown() {
     if (managedExecutor) {