Retry on configurable exception

docs/apidiffs/current_vs_latest/opentelemetry-sdk-common.txt

@@ -1,2 +1,13 @@
 Comparing source compatibility of opentelemetry-sdk-common-1.47.0-SNAPSHOT.jar against opentelemetry-sdk-common-1.46.0.jar
-No changes.
++++  NEW CLASS: PUBLIC(+) io.opentelemetry.sdk.common.export.DefaultRetryExceptionPredicate  (not serializable)
+	+++  CLASS FILE FORMAT VERSION: 52.0 <- n.a.
+	+++  NEW INTERFACE: java.util.function.Predicate
+	+++  NEW SUPERCLASS: java.lang.Object
+	+++  NEW CONSTRUCTOR: PUBLIC(+) DefaultRetryExceptionPredicate()
+	+++  NEW METHOD: PUBLIC(+) boolean test(java.io.IOException)
+**** MODIFIED CLASS: PUBLIC ABSTRACT io.opentelemetry.sdk.common.export.RetryPolicy  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++* NEW METHOD: PUBLIC(+) ABSTRACT(+) java.util.function.Predicate<java.io.IOException> getRetryExceptionPredicate()
+**** MODIFIED CLASS: PUBLIC ABSTRACT STATIC io.opentelemetry.sdk.common.export.RetryPolicy$RetryPolicyBuilder  (not serializable)
+	===  CLASS FILE FORMAT VERSION: 52.0 <- 52.0
+	+++* NEW METHOD: PUBLIC(+) ABSTRACT(+) io.opentelemetry.sdk.common.export.RetryPolicy$RetryPolicyBuilder setRetryExceptionPredicate(java.util.function.Predicate<java.io.IOException>)

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

@@ -1000,7 +1000,7 @@ void stringRepresentation() throws IOException, CertificateEncodingException {
                   + ", "
                   + "compressorEncoding=gzip, "
                   + "headers=Headers\\{.*foo=OBFUSCATED.*\\}, "
-                  + "retryPolicy=RetryPolicy\\{maxAttempts=2, initialBackoff=PT0\\.05S, maxBackoff=PT3S, backoffMultiplier=1\\.3\\}"
+                  + "retryPolicy=RetryPolicy\\{maxAttempts=2, initialBackoff=PT0\\.05S, maxBackoff=PT3S, backoffMultiplier=1\\.3, retryExceptionPredicate=io.opentelemetry.sdk.common.export.DefaultRetryExceptionPredicate.*\\}"
                   + ".*" // Maybe additional grpcChannel field, signal specific fields
                   + "\\}");
     } finally {

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

@@ -904,7 +904,7 @@ void stringRepresentation() throws IOException, CertificateEncodingException {
                   + ", "
                   + "exportAsJson=false, "
                   + "headers=Headers\\{.*foo=OBFUSCATED.*\\}, "
-                  + "retryPolicy=RetryPolicy\\{maxAttempts=2, initialBackoff=PT0\\.05S, maxBackoff=PT3S, backoffMultiplier=1\\.3\\}"
+                  + "retryPolicy=RetryPolicy\\{maxAttempts=2, initialBackoff=PT0\\.05S, maxBackoff=PT3S, backoffMultiplier=1\\.3, retryExceptionPredicate=io.opentelemetry.sdk.common.export.DefaultRetryExceptionPredicate.*\\}"
                   + ".*" // Maybe additional signal specific fields
                   + "\\}");
     } finally {

exporters/sender/okhttp/src/main/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptor.java

@@ -42,7 +42,7 @@ public RetryInterceptor(RetryPolicy retryPolicy, Function<Response, Boolean> isR
     this(
         retryPolicy,
         isRetryable,
-        RetryInterceptor::isRetryableException,
+        e -> retryPolicy.getRetryExceptionPredicate().test(e),
         TimeUnit.NANOSECONDS::sleep,
         bound -> ThreadLocalRandom.current().nextLong(bound));
   }
@@ -144,6 +144,11 @@ private static String responseStringRepresentation(Response response) {
     return joiner.toString();
   }
 
+  // Visible for testing
+  boolean shouldRetryOnException(IOException e) {
+    return isRetryableException.apply(e);
+  }
+
   // Visible for testing
   static boolean isRetryableException(IOException e) {
     if (e instanceof SocketTimeoutException) {

exporters/sender/okhttp/src/test/java/io/opentelemetry/exporter/sender/okhttp/internal/RetryInterceptorTest.java

@@ -21,14 +21,16 @@
 import com.linecorp.armeria.common.HttpResponse;
 import com.linecorp.armeria.common.HttpStatus;
 import com.linecorp.armeria.testing.junit5.server.mock.MockWebServerExtension;
+import io.opentelemetry.sdk.common.export.DefaultRetryExceptionPredicate;
 import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.io.IOException;
 import java.net.ConnectException;
+import java.net.HttpRetryException;
 import java.net.ServerSocket;
 import java.net.SocketTimeoutException;
 import java.time.Duration;
 import java.util.concurrent.TimeUnit;
-import java.util.function.Function;
+import java.util.function.Predicate;
 import okhttp3.OkHttpClient;
 import okhttp3.Request;
 import okhttp3.Response;
@@ -48,32 +50,40 @@ class RetryInterceptorTest {
 
   @Mock private RetryInterceptor.Sleeper sleeper;
   @Mock private RetryInterceptor.BoundedLongGenerator random;
-  private Function<IOException, Boolean> isRetryableException;
+  private Predicate<IOException> retryPredicate;
 
   private RetryInterceptor retrier;
   private OkHttpClient client;
 
   @BeforeEach
   void setUp() {
-    // Note: cannot replace this with lambda or method reference because we need to spy on it
-    isRetryableException =
+    DefaultRetryExceptionPredicate defaultRetryExceptionPredicate =
+        new DefaultRetryExceptionPredicate();
+    retryPredicate =
         spy(
-            new Function<IOException, Boolean>() {
+            new Predicate<IOException>() {
               @Override
-              public Boolean apply(IOException exception) {
-                return RetryInterceptor.isRetryableException(exception);
+              public boolean test(IOException e) {
+                return defaultRetryExceptionPredicate.test(e)
+                    || (e instanceof HttpRetryException
+                        && e.getMessage().contains("timeout retry"));
               }
             });
+
+    RetryPolicy retryPolicy =
+        RetryPolicy.builder()
+            .setBackoffMultiplier(1.6)
+            .setInitialBackoff(Duration.ofSeconds(1))
+            .setMaxBackoff(Duration.ofSeconds(2))
+            .setMaxAttempts(5)
+            .setRetryExceptionPredicate(retryPredicate)
+            .build();
+
     retrier =
         new RetryInterceptor(
-            RetryPolicy.builder()
-                .setBackoffMultiplier(1.6)
-                .setInitialBackoff(Duration.ofSeconds(1))
-                .setMaxBackoff(Duration.ofSeconds(2))
-                .setMaxAttempts(5)
-                .build(),
+            retryPolicy,
             r -> !r.isSuccessful(),
-            isRetryableException,
+            e -> retryPolicy.getRetryExceptionPredicate().test(e),
             sleeper,
             random);
     client = new OkHttpClient.Builder().addInterceptor(retrier).build();
@@ -154,7 +164,7 @@ void connectTimeout() throws Exception {
                 client.newCall(new Request.Builder().url("http://10.255.255.1").build()).execute())
         .isInstanceOf(SocketTimeoutException.class);
 
-    verify(isRetryableException, times(5)).apply(any());
+    verify(retryPredicate, times(5)).test(any());
     // Should retry maxAttempts, and sleep maxAttempts - 1 times
     verify(sleeper, times(4)).sleep(anyLong());
   }
@@ -174,7 +184,7 @@ void connectException() throws Exception {
                     .execute())
         .isInstanceOfAny(ConnectException.class, SocketTimeoutException.class);
 
-    verify(isRetryableException, times(5)).apply(any());
+    verify(retryPredicate, times(5)).test(any());
     // Should retry maxAttempts, and sleep maxAttempts - 1 times
     verify(sleeper, times(4)).sleep(anyLong());
   }
@@ -190,16 +200,16 @@ private static int freePort() {
   @Test
   void nonRetryableException() throws InterruptedException {
     client = connectTimeoutClient();
-    // Override isRetryableException so that no exception is retryable
-    when(isRetryableException.apply(any())).thenReturn(false);
+    // Override retryPredicate so that no exception is retryable
+    when(retryPredicate.test(any())).thenReturn(false);
 
     // Connecting to a non-routable IP address to trigger connection timeout
     assertThatThrownBy(
             () ->
                 client.newCall(new Request.Builder().url("http://10.255.255.1").build()).execute())
         .isInstanceOf(SocketTimeoutException.class);
 
-    verify(isRetryableException, times(1)).apply(any());
+    verify(retryPredicate, times(1)).test(any());
     verify(sleeper, never()).sleep(anyLong());
   }
 
@@ -214,20 +224,51 @@ private OkHttpClient connectTimeoutClient() {
   void isRetryableException() {
     // Should retry on connection timeouts, where error message is "Connect timed out" or "connect
     // timed out"
-    assertThat(
-            RetryInterceptor.isRetryableException(new SocketTimeoutException("Connect timed out")))
+    assertThat(retrier.shouldRetryOnException(new SocketTimeoutException("Connect timed out")))
         .isTrue();
-    assertThat(
-            RetryInterceptor.isRetryableException(new SocketTimeoutException("connect timed out")))
+    assertThat(retrier.shouldRetryOnException(new SocketTimeoutException("connect timed out")))
         .isTrue();
     // Shouldn't retry on read timeouts, where error message is "Read timed out"
-    assertThat(RetryInterceptor.isRetryableException(new SocketTimeoutException("Read timed out")))
+    assertThat(retrier.shouldRetryOnException(new SocketTimeoutException("Read timed out")))
         .isFalse();
-    // Shouldn't retry on write timeouts, where error message is "timeout", or other IOException
-    assertThat(RetryInterceptor.isRetryableException(new SocketTimeoutException("timeout")))
+    // Shouldn't retry on write timeouts or other IOException
+    assertThat(retrier.shouldRetryOnException(new SocketTimeoutException("timeout"))).isTrue();
+    assertThat(retrier.shouldRetryOnException(new SocketTimeoutException())).isTrue();
+    assertThat(retrier.shouldRetryOnException(new IOException("error"))).isFalse();
+
+    // Testing configured predicate
+    assertThat(retrier.shouldRetryOnException(new HttpRetryException("error", 400))).isFalse();
+    assertThat(retrier.shouldRetryOnException(new HttpRetryException("timeout retry", 400)))
+        .isTrue();
+  }
+
+  @Test
+  void isRetryableExceptionDefaultBehaviour() {
+    RetryInterceptor retryInterceptor =
+        new RetryInterceptor(RetryPolicy.getDefault(), OkHttpHttpSender::isRetryable);
+    assertThat(
+            retryInterceptor.shouldRetryOnException(
+                new SocketTimeoutException("Connect timed out")))
+        .isTrue();
+    assertThat(retryInterceptor.shouldRetryOnException(new IOException("Connect timed out")))
+        .isFalse();
+  }
+
+  @Test
+  void isRetryableExceptionCustomRetryPredicate() {
+    RetryInterceptor retryInterceptor =
+        new RetryInterceptor(
+            RetryPolicy.builder()
+                .setRetryExceptionPredicate((IOException e) -> e.getMessage().equals("retry"))
+                .build(),
+            OkHttpHttpSender::isRetryable);
+
+    assertThat(retryInterceptor.shouldRetryOnException(new IOException("some message"))).isFalse();
+    assertThat(retryInterceptor.shouldRetryOnException(new IOException("retry"))).isTrue();
+    assertThat(
+            retryInterceptor.shouldRetryOnException(
+                new SocketTimeoutException("Connect timed out")))
         .isFalse();
-    assertThat(RetryInterceptor.isRetryableException(new SocketTimeoutException())).isTrue();
-    assertThat(RetryInterceptor.isRetryableException(new IOException("error"))).isFalse();
   }
 
   private Response sendRequest() throws IOException {

sdk/common/src/main/java/io/opentelemetry/sdk/common/export/DefaultRetryExceptionPredicate.java

@@ -0,0 +1,33 @@
+/*
+ * Copyright The OpenTelemetry Authors
+ * SPDX-License-Identifier: Apache-2.0
+ */
+
+package io.opentelemetry.sdk.common.export;
+
+import java.io.IOException;
+import java.net.ConnectException;
+import java.net.SocketTimeoutException;
+import java.util.Locale;
+import java.util.function.Predicate;
+
+public class DefaultRetryExceptionPredicate implements Predicate<IOException> {
+  @Override
+  public boolean test(IOException e) {
+    if (e instanceof SocketTimeoutException) {
+      String message = e.getMessage();
+      // Connect timeouts can produce SocketTimeoutExceptions with no message, or with "connect
+      // timed out", or timeout
+      if (message == null) {
+        return true;
+      }
+      message = message.toLowerCase(Locale.ROOT);
+      return message.contains("connect timed out") || message.contains("timeout");
+    } else if (e instanceof ConnectException) {
+      // Exceptions resemble: java.net.ConnectException: Failed to connect to
+      // localhost/[0:0:0:0:0:0:0:1]:62611
+      return true;
+    }
+    return false;
+  }
+}

sdk/common/src/main/java/io/opentelemetry/sdk/common/export/RetryPolicy.java

@@ -8,7 +8,9 @@
 import static io.opentelemetry.api.internal.Utils.checkArgument;
 
 import com.google.auto.value.AutoValue;
+import java.io.IOException;
 import java.time.Duration;
+import java.util.function.Predicate;
 
 /**
  * Configuration for exporter exponential retry policy.
@@ -28,6 +30,9 @@ public abstract class RetryPolicy {
 
   private static final double DEFAULT_BACKOFF_MULTIPLIER = 1.5;
 
+  private static final Predicate<IOException> DEFAULT_RETRY_PREDICATE =
+      new DefaultRetryExceptionPredicate();
+
   private static final RetryPolicy DEFAULT = RetryPolicy.builder().build();
 
   RetryPolicy() {}
@@ -43,7 +48,8 @@ public static RetryPolicyBuilder builder() {
         .setMaxAttempts(DEFAULT_MAX_ATTEMPTS)
         .setInitialBackoff(Duration.ofSeconds(DEFAULT_INITIAL_BACKOFF_SECONDS))
         .setMaxBackoff(Duration.ofSeconds(DEFAULT_MAX_BACKOFF_SECONDS))
-        .setBackoffMultiplier(DEFAULT_BACKOFF_MULTIPLIER);
+        .setBackoffMultiplier(DEFAULT_BACKOFF_MULTIPLIER)
+        .setRetryExceptionPredicate(DEFAULT_RETRY_PREDICATE);
   }
 
   /**
@@ -66,6 +72,9 @@ public static RetryPolicyBuilder builder() {
   /** Returns the backoff multiplier. */
   public abstract double getBackoffMultiplier();
 
+  /** Returns the predicate if exception is retryable. */
+  public abstract Predicate<IOException> getRetryExceptionPredicate();
+
   /** Builder for {@link RetryPolicy}. */
   @AutoValue.Builder
   public abstract static class RetryPolicyBuilder {
@@ -96,6 +105,13 @@ public abstract static class RetryPolicyBuilder {
      */
     public abstract RetryPolicyBuilder setBackoffMultiplier(double backoffMultiplier);
 
+    /**
+     * Set the predicate to determine if retry should happen based on exception. No retry by
+     * default.
+     */
+    public abstract RetryPolicyBuilder setRetryExceptionPredicate(
+        Predicate<IOException> retryExceptionPredicate);
+
     abstract RetryPolicy autoBuild();
 
     /** Build and return a {@link RetryPolicy} with the values of this builder. */