Retry OkHttp gRPC exports on the current thread instead of re-dispatching

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

@@ -214,8 +214,9 @@ private void sendAttempt(
               onError.accept(new SocketTimeoutException("Call timed out"));
               return;
             }
-            enqueue(
+            startCall(
                 call,
+                attempt,
                 new Callback() {
                   @Override
                   public void onFailure(Call call, IOException e) {
@@ -265,6 +266,29 @@ && isRetryable(resolvedResponse)
     }
   }
 
+  private void startCall(Call call, int attempt, Callback callback) {
+    if (attempt == 0) {
+      enqueue(call, callback);
+      return;
+    }
+    // Retries already run on a dispatcher thread, so execute them on this thread instead of
+    // enqueueing a new call, which a bounded executor would reject.
+    boolean shutdown;
+    synchronized (shutdownLock) {
+      shutdown = isShutdown;
+    }
+    if (shutdown) {
+      call.cancel();
+      callback.onFailure(call, new IOException("Canceled"));
+      return;
+    }
+    try {
+      callback.onResponse(call, call.execute());
+    } catch (IOException e) {
+      callback.onFailure(call, e);
+    }
+  }
+
   private void enqueue(Call call, Callback callback) {
     synchronized (shutdownLock) {
       if (!isShutdown) {

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

@@ -16,13 +16,16 @@
 import io.opentelemetry.sdk.common.export.GrpcResponse;
 import io.opentelemetry.sdk.common.export.GrpcStatusCode;
 import io.opentelemetry.sdk.common.export.MessageWriter;
+import io.opentelemetry.sdk.common.export.RetryPolicy;
 import java.io.IOException;
 import java.io.OutputStream;
+import java.net.ConnectException;
 import java.net.ServerSocket;
 import java.security.Security;
 import java.time.Duration;
 import java.util.Collections;
 import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -143,6 +146,47 @@ void send_rejectedExecution_callsOnError() {
     assertThat(responseRef.get()).isNull();
   }
 
+  @Test
+  void send_retryWithSingleThreadExecutor_doesNotRejectRetry() throws Exception {
+    // Allocate an ephemeral port and immediately close it so that every attempt fails to connect
+    int port;
+    try (ServerSocket socket = new ServerSocket(0)) {
+      port = socket.getLocalPort();
+    }
+    // The only thread is busy running the first attempt when the retry starts
+    ThreadPoolExecutor executor =
+        new ThreadPoolExecutor(1, 1, 0, TimeUnit.SECONDS, new SynchronousQueue<>());
+
+    OkHttpGrpcSender sender =
+        new OkHttpGrpcSender(
+            "http://localhost:" + port,
+            null,
+            Duration.ofSeconds(10),
+            Duration.ofSeconds(10),
+            Collections::emptyMap,
+            RetryPolicy.builder().setMaxAttempts(2).setInitialBackoff(Duration.ofMillis(1)).build(),
+            null,
+            null,
+            executor,
+            Long.MAX_VALUE,
+            null);
+
+    CompletableFuture<Throwable> error = new CompletableFuture<>();
+    try {
+      sender.send(
+          new TestMessageWriter(),
+          response -> error.completeExceptionally(new AssertionError("Unexpected response")),
+          error::complete);
+
+      // Without running the retry on the current thread, the retry is rejected by the executor and
+      // fails with "executor rejected" instead of the connection error
+      assertThat(error.get(10, TimeUnit.SECONDS)).isInstanceOf(ConnectException.class);
+    } finally {
+      sender.shutdown();
+      executor.shutdownNow();
+    }
+  }
+
   @Test
   void shutdown_CompletableResultCodeShouldWaitForThreads() throws Exception {
     // This test verifies that shutdown() returns a CompletableResultCode that only