diff --git a/exporters/sender/grpc-managed-channel/src/main/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSender.java b/exporters/sender/grpc-managed-channel/src/main/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSender.java index 3a755be609f..1a2d75d8af4 100644 --- a/exporters/sender/grpc-managed-channel/src/main/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSender.java +++ b/exporters/sender/grpc-managed-channel/src/main/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSender.java @@ -36,8 +36,11 @@ import java.util.Map; import java.util.concurrent.Executor; import java.util.concurrent.ExecutorService; +import java.util.concurrent.TimeUnit; import java.util.function.Consumer; import java.util.function.Supplier; +import java.util.logging.Level; +import java.util.logging.Logger; import javax.annotation.Nullable; /** @@ -48,6 +51,8 @@ */ public final class UpstreamGrpcSender implements GrpcSender { + private static final Logger logger = Logger.getLogger(UpstreamGrpcSender.class.getName()); + private static final MethodDescriptor.Marshaller REQUEST_MARSHALER = new MethodDescriptor.Marshaller() { @Override @@ -231,8 +236,35 @@ private static Status fromThrowable(Throwable cause) { @Override public CompletableResultCode shutdown() { if (shutdownChannel) { + // Use shutdownNow() to cancel in flight calls immediately channel.shutdownNow(); + + // Wait for the channel to terminate in a background thread + CompletableResultCode result = new CompletableResultCode(); + Thread terminationThread = + new Thread( + () -> { + try { + // Wait up to 5 seconds for the channel to terminate + // Even if timeout occurs, we succeed since these are daemon threads + boolean terminated = channel.awaitTermination(5, TimeUnit.SECONDS); + if (!terminated) { + logger.log( + Level.WARNING, + "Channel did not terminate within 5 seconds, proceeding with shutdown since threads are daemon threads."); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + result.succeed(); + } + }, + "grpc-shutdown"); + terminationThread.setDaemon(true); + terminationThread.start(); + return result; } + return CompletableResultCode.ofSuccess(); } } diff --git a/exporters/sender/grpc-managed-channel/src/test/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSenderTest.java b/exporters/sender/grpc-managed-channel/src/test/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSenderTest.java new file mode 100644 index 00000000000..8204d3cddec --- /dev/null +++ b/exporters/sender/grpc-managed-channel/src/test/java/io/opentelemetry/exporter/sender/grpc/managedchannel/internal/UpstreamGrpcSenderTest.java @@ -0,0 +1,146 @@ +/* + * Copyright The OpenTelemetry Authors + * SPDX-License-Identifier: Apache-2.0 + */ + +package io.opentelemetry.exporter.sender.grpc.managedchannel.internal; + +import static org.assertj.core.api.Assertions.assertThat; + +import io.grpc.CallOptions; +import io.grpc.ClientCall; +import io.grpc.ManagedChannel; +import io.grpc.MethodDescriptor; +import io.opentelemetry.sdk.common.CompletableResultCode; +import java.time.Duration; +import java.util.Collections; +import java.util.concurrent.CountDownLatch; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; +import org.junit.jupiter.api.Test; + +class UpstreamGrpcSenderTest { + + @Test + void shutdown_ManagedChannel_WaitsForChannelTermination() throws Exception { + // The result code must only complete after the channel terminates, not as soon as + // shutdownNow() returns. + FakeManagedChannel channel = new FakeManagedChannel(/* terminates= */ true); + UpstreamGrpcSender sender = newSender(channel, /* shutdownChannel= */ true); + + CompletableResultCode result = sender.shutdown(); + + assertThat(channel.awaitTerminationEntered.await(10, TimeUnit.SECONDS)).isTrue(); + assertThat(channel.shutdownNowCalled()).isTrue(); + assertThat(result.isDone()) + .describedAs("result should not complete before the channel terminates") + .isFalse(); + + channel.releaseAwaitTermination.countDown(); + + assertThat(result.join(10, TimeUnit.SECONDS).isSuccess()).isTrue(); + } + + @Test + void shutdown_UnmanagedChannel_ReturnsImmediatelyAndLeavesChannelOpen() { + // A caller supplied channel is not ours to close. + FakeManagedChannel channel = new FakeManagedChannel(/* terminates= */ true); + UpstreamGrpcSender sender = newSender(channel, /* shutdownChannel= */ false); + + CompletableResultCode result = sender.shutdown(); + + assertThat(result.isDone()).isTrue(); + assertThat(result.isSuccess()).isTrue(); + assertThat(channel.shutdownNowCalled()).isFalse(); + assertThat(channel.awaitTerminationCalled()).isFalse(); + } + + @Test + void shutdown_ChannelDoesNotTerminateInTime_Succeeds() { + // Termination timing out must not hang or fail the shutdown. + FakeManagedChannel channel = new FakeManagedChannel(/* terminates= */ false); + channel.releaseAwaitTermination.countDown(); + UpstreamGrpcSender sender = newSender(channel, /* shutdownChannel= */ true); + + CompletableResultCode result = sender.shutdown(); + + assertThat(result.join(10, TimeUnit.SECONDS).isSuccess()).isTrue(); + } + + private static UpstreamGrpcSender newSender(ManagedChannel channel, boolean shutdownChannel) { + return new UpstreamGrpcSender( + channel, + "opentelemetry.proto.collector.trace.v1.TraceService/Export", + /* compressor= */ null, + shutdownChannel, + Duration.ofSeconds(10), + Collections::emptyMap, + /* executorService= */ null); + } + + /** + * A {@link ManagedChannel} which records shutdown interactions and lets the test control when + * {@link #awaitTermination(long, TimeUnit)} returns. + */ + private static class FakeManagedChannel extends ManagedChannel { + + final CountDownLatch awaitTerminationEntered = new CountDownLatch(1); + final CountDownLatch releaseAwaitTermination = new CountDownLatch(1); + + private final boolean terminates; + private final AtomicBoolean shutdownNow = new AtomicBoolean(); + private final AtomicBoolean awaitTermination = new AtomicBoolean(); + + FakeManagedChannel(boolean terminates) { + this.terminates = terminates; + } + + boolean shutdownNowCalled() { + return shutdownNow.get(); + } + + boolean awaitTerminationCalled() { + return awaitTermination.get(); + } + + @Override + public ManagedChannel shutdown() { + return this; + } + + @Override + public ManagedChannel shutdownNow() { + shutdownNow.set(true); + return this; + } + + @Override + public boolean isShutdown() { + return shutdownNow.get(); + } + + @Override + public boolean isTerminated() { + return terminates && shutdownNow.get(); + } + + @Override + public boolean awaitTermination(long timeout, TimeUnit unit) throws InterruptedException { + awaitTermination.set(true); + awaitTerminationEntered.countDown(); + releaseAwaitTermination.await(10, TimeUnit.SECONDS); + return terminates; + } + + @Override + public ClientCall newCall( + MethodDescriptor methodDescriptor, CallOptions callOptions) { + throw new UnsupportedOperationException(); + } + + @Override + public String authority() { + return "fake"; + } + } +}