From c8618617cbc89047fc58c8484e1e761ffe219fa3 Mon Sep 17 00:00:00 2001 From: Eunbin Son Date: Fri, 17 Jul 2026 10:43:09 +0900 Subject: [PATCH] Standardize JdkHttpSender shutdown to await executor termination --- .../sender/jdk/internal/JdkHttpSender.java | 41 ++++++++++++++++++- .../jdk/internal/JdkHttpSenderTest.java | 39 ++++++++++++++++++ 2 files changed, 78 insertions(+), 2 deletions(-) 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..99ebdb68013 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 @@ -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(); 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..db245a23a77 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 @@ -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()))