Standardize JdkHttpSender shutdown to await executor termination

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

@@ -414,9 +414,46 @@ private void resetPool() {
 
   @Override
   public CompletableResultCode shutdown() {
-    if (managedExecutor) {
-      executorService.shutdown();
+    if (!managedExecutor) {
+      return closeClient();
     }
+
+    // Use shutdownNow() to interrupt in-flight requests, including retry backoff sleeps
+    executorService.shutdownNow();
+
+    // Wait for threads to terminate in a background thread. Closing the client also blocks until
+    // in-flight requests complete, so it must not run on the caller's thread either.
+    CompletableResultCode result = new CompletableResultCode();
+    Thread terminationThread =
+        new Thread(
+            () -> {
+              try {
+                // Wait up to 5 seconds for threads to terminate
+                // Even if timeout occurs, we proceed since these are daemon threads
+                boolean terminated = executorService.awaitTermination(5, TimeUnit.SECONDS);
+                if (!terminated) {
+                  logger.log(
+                      Level.WARNING,
+                      "Executor did not terminate within 5 seconds, proceeding with shutdown since threads are daemon threads.");
+                }
+              } catch (InterruptedException e) {
+                Thread.currentThread().interrupt();
+              } finally {
+                CompletableResultCode closeResult = closeClient();
+                if (closeResult.isSuccess()) {
+                  result.succeed();
+                } else {
+                  result.failExceptionally(closeResult.getFailureThrowable());
+                }
+              }
+            },
+            "jdkhttp-shutdown");
+    terminationThread.setDaemon(true);
+    terminationThread.start();
+    return result;
+  }
+
+  private CompletableResultCode closeClient() {
     if (AutoCloseable.class.isInstance(client)) {
       try {
         AutoCloseable.class.cast(client).close();

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

@@ -34,6 +34,8 @@
 import java.time.Duration;
 import java.util.Collections;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.RejectedExecutionException;
 import java.util.concurrent.SynchronousQueue;
 import java.util.concurrent.ThreadPoolExecutor;
@@ -108,6 +110,43 @@ void testShutdownException() throws Exception {
     assertThat(result.getFailureThrowable().getMessage()).isEqualTo("testShutdownException");
   }
 
+  @Test
+  void shutdown_managedExecutor_awaitsTermination() {
+    CompletableResultCode result = sender.shutdown();
+    result.join(10, TimeUnit.SECONDS);
+
+    assertThat(result.isSuccess()).isTrue();
+    assertThat(sender)
+        .extracting("executorService", as(InstanceOfAssertFactories.type(ExecutorService.class)))
+        .satisfies(executor -> assertThat(executor.isTerminated()).isTrue());
+  }
+
+  @Test
+  void shutdown_nonManagedExecutor_doesNotShutDownExecutor() {
+    ExecutorService customExecutor = Executors.newSingleThreadExecutor();
+    try {
+      JdkHttpSender testSender =
+          new JdkHttpSender(
+              mockHttpClient,
+              URI.create("http://localhost"),
+              "text/plain",
+              null,
+              Duration.ofSeconds(10),
+              Collections::emptyMap,
+              null,
+              customExecutor,
+              Long.MAX_VALUE);
+
+      CompletableResultCode result = testSender.shutdown();
+
+      assertThat(result.isDone()).isTrue();
+      assertThat(result.isSuccess()).isTrue();
+      assertThat(customExecutor.isShutdown()).isFalse();
+    } finally {
+      customExecutor.shutdownNow();
+    }
+  }
+
   @Test
   void sendInternal_RetryableConnectTimeoutException() throws IOException, InterruptedException {
     assertThatThrownBy(() -> sender.sendInternal(new NoOpRequestBodyWriter()))