Bound JdkHttpSender thread pool to prevent DoS via unbounded thread creation

The default executor used Integer.MAX_VALUE max threads with a SynchronousQueue, allowing thousands of threads under burst load. Cap at max(availableProcessors, 5) with CallerRunsPolicy for backpressure, and await termination on shutdown so in-flight requests complete before the HttpClient is closed.

Author: abdessattar23

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

@@ -31,6 +31,7 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentLinkedQueue;
 import java.util.concurrent.ExecutorService;
+import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadLocalRandom;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -133,7 +134,7 @@ public final class JdkHttpSender implements HttpSender {
   private static ExecutorService newExecutor() {
     return new ThreadPoolExecutor(
         0,
-        Integer.MAX_VALUE,
+        Math.max(Runtime.getRuntime().availableProcessors(), 5),
         60,
         TimeUnit.SECONDS,
         new SynchronousQueue<>(),
@@ -157,24 +158,28 @@ private static HttpClient configureClient(
   @Override
   public void send(
       MessageWriter messageWriter, Consumer<HttpResponse> onResponse, Consumer<Throwable> onError) {
-    CompletableFuture<HttpResponse> unused =
-        CompletableFuture.supplyAsync(
-                () -> {
-                  try {
-                    return sendInternal(messageWriter);
-                  } catch (IOException e) {
-                    throw new UncheckedIOException(e);
-                  }
-                },
-                executorService)
-            .whenComplete(
-                (httpResponse, throwable) -> {
-                  if (throwable != null) {
-                    onError.accept(throwable);
-                    return;
-                  }
-                  onResponse.accept(httpResponse);
-                });
+    try {
+      CompletableFuture<HttpResponse> unused =
+          CompletableFuture.supplyAsync(
+                  () -> {
+                    try {
+                      return sendInternal(messageWriter);
+                    } catch (IOException e) {
+                      throw new UncheckedIOException(e);
+                    }
+                  },
+                  executorService)
+              .whenComplete(
+                  (httpResponse, throwable) -> {
+                    if (throwable != null) {
+                      onError.accept(throwable);
+                      return;
+                    }
+                    onResponse.accept(httpResponse);
+                  });
+    } catch (RejectedExecutionException e) {
+      onError.accept(e);
+    }
   }
 
   // Visible for testing
@@ -409,6 +414,11 @@ private void resetPool() {
   public CompletableResultCode shutdown() {
     if (managedExecutor) {
       executorService.shutdown();
+      try {
+        executorService.awaitTermination(10, TimeUnit.SECONDS);
+      } catch (InterruptedException e) {
+        Thread.currentThread().interrupt();
+      }
     }
     if (AutoCloseable.class.isInstance(client)) {
       try {

exporters/sender/jdk/src/test/java/io/opentelemetry/exporter/sender/jdk/internal/JdkHttpSenderTest.java

@@ -27,6 +27,7 @@
 import java.net.http.HttpConnectTimeoutException;
 import java.time.Duration;
 import java.util.Collections;
+import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import javax.net.ssl.SSLException;
 import org.assertj.core.api.InstanceOfAssertFactories;
@@ -166,6 +167,38 @@ void sendInternal_NonRetryableException() throws IOException, InterruptedExcepti
     verify(mockHttpClient, times(1)).send(any(), any());
   }
 
+  @Test
+  void defaultExecutor_isBounded() {
+    JdkHttpSender defaultSender =
+        new JdkHttpSender(
+            URI.create("http://localhost"),
+            "text/plain",
+            null,
+            Duration.ofNanos(1),
+            Duration.ofSeconds(10),
+            Collections::emptyMap,
+            null,
+            null,
+            null,
+            null,
+            Long.MAX_VALUE);
+
+    try {
+      int expectedMax = Math.max(Runtime.getRuntime().availableProcessors(), 5);
+      assertThat(defaultSender)
+          .extracting(
+              "executorService", as(InstanceOfAssertFactories.type(ThreadPoolExecutor.class)))
+          .satisfies(
+              executor -> {
+                assertThat(executor.getMaximumPoolSize()).isEqualTo(expectedMax);
+                assertThat(executor.getRejectedExecutionHandler())
+                    .isInstanceOf(ThreadPoolExecutor.AbortPolicy.class);
+              });
+    } finally {
+      defaultSender.shutdown();
+    }
+  }
+
   @Test
   void connectTimeout() {
     sender =