diff --git a/exporters/common/src/main/java/io/opentelemetry/exporter/internal/RetryUtil.java b/exporters/common/src/main/java/io/opentelemetry/exporter/internal/RetryUtil.java index eab9e30cab2..ea188d2602e 100644 --- a/exporters/common/src/main/java/io/opentelemetry/exporter/internal/RetryUtil.java +++ b/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 retryableGrpcStatusCodes() { public static Set 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(); + } + } } diff --git a/exporters/common/src/test/java/io/opentelemetry/exporter/internal/RetryUtilTest.java b/exporters/common/src/test/java/io/opentelemetry/exporter/internal/RetryUtilTest.java new file mode 100644 index 00000000000..4fd1d35901c --- /dev/null +++ b/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(); + } +} diff --git a/exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/AbstractHttpTelemetryExporterTest.java b/exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/AbstractHttpTelemetryExporterTest.java index 5f5b13b1adf..78f798f8a83 100644 --- a/exporters/otlp/testing-internal/src/main/java/io/opentelemetry/exporter/otlp/testing/internal/AbstractHttpTelemetryExporterTest.java +++ b/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( diff --git a/exporters/sender/jdk/src/main/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSender.java b/exporters/sender/jdk/src/main/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSender.java index 6c3610e7345..d203d39ece8 100644 --- a/exporters/sender/jdk/src/main/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSender.java +++ b/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 retryableStatusCodes = Set.of(429, 502, 503, 504); + private static final Set retryableStatusCodes = RetryUtil.retryableHttpResponseCodes(); private static final ThreadLocal threadLocalBaos = ThreadLocal.withInitial(NoCopyByteArrayOutputStream::new); @@ -74,6 +76,8 @@ public final class JdkHttpSender implements HttpSender { @Nullable private final RetryPolicy retryPolicy; private final Predicate retryExceptionPredicate; private final long maxResponseBodySize; + private final Sleeper sleeper; + private final Supplier 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>> headerSupplier, + @Nullable RetryPolicy retryPolicy, + @Nullable ExecutorService executorService, + long maxResponseBodySize, + Sleeper sleeper, + Supplier 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 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 sendRequest( HttpRequest.Builder requestBuilder, ByteBufferPool byteBufferPool) throws IOException { try { - java.net.http.HttpResponse 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) { diff --git a/exporters/sender/jdk/src/test/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSenderTest.java b/exporters/sender/jdk/src/test/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSenderTest.java index e329266b80d..b0b6a306f04 100644 --- a/exporters/sender/jdk/src/test/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSenderTest.java +++ b/exporters/sender/jdk/src/test/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSenderTest.java @@ -39,6 +39,7 @@ import java.util.concurrent.ThreadPoolExecutor; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicReference; +import java.util.function.Supplier; import javax.net.ssl.SSLException; import org.assertj.core.api.InstanceOfAssertFactories; import org.junit.jupiter.api.BeforeEach; @@ -57,6 +58,8 @@ class JdkHttpSenderTest { private final HttpClient realHttpClient = HttpClient.newBuilder().connectTimeout(Duration.ofMillis(10)).build(); @Mock private HttpClient mockHttpClient; + @Mock private JdkHttpSender.Sleeper sleeper; + @Mock private Supplier randomJitter; private JdkHttpSender sender; @BeforeEach @@ -166,6 +169,89 @@ void sendInternal_RetryableIoException() throws IOException, InterruptedExceptio verify(mockHttpClient, times(2)).send(any(), any()); } + @SuppressWarnings("unchecked") + @Test + void sendInternal_RetryableResponse_UsesRetryAfterDelay() + throws IOException, InterruptedException { + java.net.http.HttpResponse throttledResponse = + mock(java.net.http.HttpResponse.class); + when(throttledResponse.statusCode()).thenReturn(429); + when(throttledResponse.body()).thenReturn(new ByteArrayInputStream(new byte[0])); + when(throttledResponse.headers()) + .thenReturn( + HttpHeaders.of( + Collections.singletonMap("Retry-After", Collections.singletonList("30")), + (a, b) -> true)); + + java.net.http.HttpResponse okResponse = mock(java.net.http.HttpResponse.class); + when(okResponse.statusCode()).thenReturn(200); + when(okResponse.body()).thenReturn(new ByteArrayInputStream(new byte[0])); + when(okResponse.headers()).thenReturn(HttpHeaders.of(Collections.emptyMap(), (a, b) -> true)); + + doReturn(throttledResponse, okResponse).when(mockHttpClient).send(any(), any()); + + sender = + new JdkHttpSender( + mockHttpClient, + URI.create("http://localhost"), + "text/plain", + null, + Duration.ofSeconds(10), + Collections::emptyMap, + RetryPolicy.builder().setMaxAttempts(2).setInitialBackoff(Duration.ofMillis(1)).build(), + null, + Long.MAX_VALUE, + sleeper, + randomJitter); + + HttpResponse response = sender.sendInternal(new NoOpRequestBodyWriter()); + + assertThat(response.getStatusCode()).isEqualTo(200); + verify(sleeper).sleep(TimeUnit.SECONDS.toNanos(30)); + } + + @SuppressWarnings("unchecked") + @Test + void sendInternal_RetryableResponse_MalformedRetryAfterFallsBack() + throws IOException, InterruptedException { + java.net.http.HttpResponse throttledResponse = + mock(java.net.http.HttpResponse.class); + when(throttledResponse.statusCode()).thenReturn(503); + when(throttledResponse.body()).thenReturn(new ByteArrayInputStream(new byte[0])); + when(throttledResponse.headers()) + .thenReturn( + HttpHeaders.of( + Collections.singletonMap("Retry-After", Collections.singletonList("bad-value")), + (a, b) -> true)); + + java.net.http.HttpResponse okResponse = mock(java.net.http.HttpResponse.class); + when(okResponse.statusCode()).thenReturn(200); + when(okResponse.body()).thenReturn(new ByteArrayInputStream(new byte[0])); + when(okResponse.headers()).thenReturn(HttpHeaders.of(Collections.emptyMap(), (a, b) -> true)); + + doReturn(throttledResponse, okResponse).when(mockHttpClient).send(any(), any()); + when(randomJitter.get()).thenReturn(1.0d); + + sender = + new JdkHttpSender( + mockHttpClient, + URI.create("http://localhost"), + "text/plain", + null, + Duration.ofSeconds(10), + Collections::emptyMap, + RetryPolicy.builder().setMaxAttempts(2).setInitialBackoff(Duration.ofMillis(1)).build(), + null, + Long.MAX_VALUE, + sleeper, + randomJitter); + + HttpResponse response = sender.sendInternal(new NoOpRequestBodyWriter()); + + assertThat(response.getStatusCode()).isEqualTo(200); + verify(sleeper).sleep(TimeUnit.MILLISECONDS.toNanos(1)); + } + @Test void sendInternal_NonRetryableException() throws IOException, InterruptedException { doThrow(new SSLException("unknown error")).when(mockHttpClient).send(any(), any()); diff --git a/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpHttpSender.java b/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpHttpSender.java index b5c611d1bcb..822ba376b30 100644 --- a/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpHttpSender.java +++ b/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/OkHttpHttpSender.java @@ -20,6 +20,7 @@ import java.util.Collections; import java.util.List; import java.util.Map; +import java.util.OptionalLong; import java.util.concurrent.ExecutorService; import java.util.concurrent.RejectedExecutionException; import java.util.concurrent.TimeUnit; @@ -103,7 +104,9 @@ public OkHttpHttpSender( } if (retryPolicy != null) { - builder.addInterceptor(new RetryInterceptor(retryPolicy, OkHttpHttpSender::isRetryable)); + builder.addInterceptor( + new RetryInterceptor( + retryPolicy, OkHttpHttpSender::isRetryable, OkHttpHttpSender::retryDelayNanos)); } boolean isPlainHttp = endpoint.getScheme().equals("http"); @@ -121,6 +124,13 @@ public OkHttpHttpSender( this.maxResponseBodySize = maxResponseBodySize; } + private static OptionalLong retryDelayNanos(Response response) { + if (!RetryUtil.isOtlpHttpThrottleStatusCode(response.code())) { + return OptionalLong.empty(); + } + return RetryUtil.retryAfterNanos(response.header("Retry-After")); + } + @Override public void send( MessageWriter messageWriter, Consumer onResponse, Consumer onError) { diff --git a/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java b/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java index 988c8277c26..604f172ee40 100644 --- a/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java +++ b/exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java @@ -13,6 +13,7 @@ import java.net.SocketException; import java.net.SocketTimeoutException; import java.net.UnknownHostException; +import java.util.OptionalLong; import java.util.StringJoiner; import java.util.concurrent.ThreadLocalRandom; import java.util.concurrent.TimeUnit; @@ -36,12 +37,21 @@ public final class RetryInterceptor implements Interceptor { private final RetryPolicy retryPolicy; private final Function isRetryable; + private final Function retryDelayNanosExtractor; private final Predicate retryExceptionPredicate; private final Sleeper sleeper; private final Supplier randomJitter; /** Constructs a new retrier. */ public RetryInterceptor(RetryPolicy retryPolicy, Function isRetryable) { + this(retryPolicy, isRetryable, response -> OptionalLong.empty()); + } + + /** Constructs a new retrier with an optional server-provided retry delay override. */ + public RetryInterceptor( + RetryPolicy retryPolicy, + Function isRetryable, + Function retryDelayNanosExtractor) { this( retryPolicy, isRetryable, @@ -49,7 +59,8 @@ public RetryInterceptor(RetryPolicy retryPolicy, Function isR ? RetryInterceptor::isRetryableException : retryPolicy.getRetryExceptionPredicate(), TimeUnit.NANOSECONDS::sleep, - () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d)); + () -> ThreadLocalRandom.current().nextDouble(0.8d, 1.2d), + retryDelayNanosExtractor); } // Visible for testing @@ -59,8 +70,26 @@ public RetryInterceptor(RetryPolicy retryPolicy, Function isR Predicate retryExceptionPredicate, Sleeper sleeper, Supplier randomJitter) { + this( + retryPolicy, + isRetryable, + retryExceptionPredicate, + sleeper, + randomJitter, + response -> OptionalLong.empty()); + } + + // Visible for testing + RetryInterceptor( + RetryPolicy retryPolicy, + Function isRetryable, + Predicate retryExceptionPredicate, + Sleeper sleeper, + Supplier randomJitter, + Function retryDelayNanosExtractor) { this.retryPolicy = retryPolicy; this.isRetryable = isRetryable; + this.retryDelayNanosExtractor = retryDelayNanosExtractor; this.retryExceptionPredicate = retryExceptionPredicate; this.sleeper = sleeper; this.randomJitter = randomJitter; @@ -72,14 +101,19 @@ public Response intercept(Chain chain) throws IOException { IOException exception = null; int attempt = 0; long nextBackoffNanos = retryPolicy.getInitialBackoff().toNanos(); + OptionalLong retryDelayNanos = OptionalLong.empty(); do { if (attempt > 0) { // Compute and sleep for backoff // https://github.com/grpc/proposal/blob/master/A6-client-retries.md#exponential-backoff long currentBackoffNanos = Math.min(nextBackoffNanos, retryPolicy.getMaxBackoff().toNanos()); - long backoffNanos = (long) (randomJitter.get() * currentBackoffNanos); + long backoffNanos = + retryDelayNanos.isPresent() + ? retryDelayNanos.getAsLong() + : (long) (randomJitter.get() * currentBackoffNanos); nextBackoffNanos = (long) (currentBackoffNanos * retryPolicy.getBackoffMultiplier()); + retryDelayNanos = OptionalLong.empty(); try { sleeper.sleep(backoffNanos); } catch (InterruptedException e) { @@ -109,6 +143,8 @@ public Response intercept(Chain chain) throws IOException { if (!retryable) { return response; } + OptionalLong serverRetryDelay = retryDelayNanosExtractor.apply(response); + retryDelayNanos = serverRetryDelay == null ? OptionalLong.empty() : serverRetryDelay; } else { throw new NullPointerException("response cannot be null."); } diff --git a/exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptorTest.java b/exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptorTest.java index eeff92abf3d..b6d8845cfad 100644 --- a/exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptorTest.java +++ b/exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptorTest.java @@ -31,6 +31,7 @@ import java.net.SocketTimeoutException; import java.net.UnknownHostException; import java.time.Duration; +import java.util.OptionalLong; import java.util.concurrent.TimeUnit; import java.util.function.Predicate; import java.util.function.Supplier; @@ -178,6 +179,35 @@ public Void answer(InvocationOnMock invocation) throws Throwable { } } + @Test + void retryDelayOverride() throws Exception { + succeedOnAttempt(2); + doNothing().when(sleeper).sleep(anyLong()); + + RetryInterceptor retryInterceptor = + new RetryInterceptor( + RetryPolicy.builder() + .setBackoffMultiplier(1.6) + .setInitialBackoff(Duration.ofSeconds(1)) + .setMaxBackoff(Duration.ofSeconds(2)) + .setMaxAttempts(5) + .setRetryExceptionPredicate(retryExceptionPredicate) + .build(), + r -> !r.isSuccessful(), + retryExceptionPredicate, + sleeper, + random, + response -> OptionalLong.of(123L)); + client = new OkHttpClient.Builder().addInterceptor(retryInterceptor).build(); + + try (Response response = sendRequest()) { + assertThat(response.isSuccessful()).isTrue(); + } + + verify(sleeper).sleep(123L); + verifyNoInteractions(random); + } + @Test void connectTimeout() throws Exception { client = connectTimeoutClient();