Add proper shutdown implementations for all exporters

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/OkHttpGrpcExporter.java

@@ -63,6 +63,7 @@ public final class OkHttpGrpcExporter<T extends Marshaler> implements GrpcExport
 
   // We only log unimplemented once since it's a configuration issue that won't be recovered.
   private final AtomicBoolean loggedUnimplemented = new AtomicBoolean();
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
 
   private final String type;
   private final ExporterMetrics exporterMetrics;
@@ -91,6 +92,10 @@ public final class OkHttpGrpcExporter<T extends Marshaler> implements GrpcExport
 
   @Override
   public CompletableResultCode export(T exportRequest, int numItems) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+
     exporterMetrics.addSeen(numItems);
 
     Request.Builder requestBuilder = new Request.Builder().url(url).headers(headers);
@@ -209,6 +214,10 @@ private static String grpcMessage(Response response) {
 
   @Override
   public CompletableResultCode shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.WARNING, "Calling shutdown() multiple times.");
+      return CompletableResultCode.ofSuccess();
+    }
     client.dispatcher().cancelAll();
     client.dispatcher().executorService().shutdownNow();
     client.connectionPool().evictAll();

exporters/common/src/main/java/io/opentelemetry/exporter/internal/grpc/UpstreamGrpcExporter.java

@@ -36,6 +36,7 @@ public final class UpstreamGrpcExporter<T extends Marshaler> implements GrpcExpo
 
   // We only log unavailable once since it's a configuration issue that won't be recovered.
   private final AtomicBoolean loggedUnimplemented = new AtomicBoolean();
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
 
   private final String type;
   private final ExporterMetrics exporterMetrics;
@@ -57,6 +58,10 @@ public final class UpstreamGrpcExporter<T extends Marshaler> implements GrpcExpo
 
   @Override
   public CompletableResultCode export(T exportRequest, int numItems) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+
     exporterMetrics.addSeen(numItems);
 
     CompletableResultCode result = new CompletableResultCode();
@@ -118,6 +123,9 @@ public void onFailure(Throwable t) {
 
   @Override
   public CompletableResultCode shutdown() {
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.WARNING, "Calling shutdown() multiple times.");
+    }
     return CompletableResultCode.ofSuccess();
   }
 }

exporters/common/src/main/java/io/opentelemetry/exporter/internal/okhttp/OkHttpExporter.java

@@ -13,6 +13,7 @@
 import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.internal.ThrottlingLogger;
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Function;
 import java.util.function.Supplier;
 import java.util.logging.Level;
@@ -44,6 +45,7 @@ public final class OkHttpExporter<T extends Marshaler> {
   private static final Logger internalLogger = Logger.getLogger(OkHttpExporter.class.getName());
 
   private final ThrottlingLogger logger = new ThrottlingLogger(internalLogger);
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
 
   private final String type;
   private final OkHttpClient client;
@@ -76,6 +78,10 @@ public final class OkHttpExporter<T extends Marshaler> {
   }
 
   public CompletableResultCode export(T exportRequest, int numItems) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+
     exporterMetrics.addSeen(numItems);
 
     Request.Builder requestBuilder = new Request.Builder().url(url);
@@ -139,11 +145,14 @@ public void onResponse(Call call, Response response) {
   }
 
   public CompletableResultCode shutdown() {
-    CompletableResultCode result = CompletableResultCode.ofSuccess();
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.WARNING, "Calling shutdown() multiple times.");
+      return CompletableResultCode.ofSuccess();
+    }
     client.dispatcher().cancelAll();
     client.dispatcher().executorService().shutdownNow();
     client.connectionPool().evictAll();
-    return result;
+    return CompletableResultCode.ofSuccess();
   }
 
   static boolean isRetryable(Response response) {

exporters/jaeger-thrift/src/main/java/io/opentelemetry/exporter/jaeger/thrift/JaegerThriftSpanExporter.java

@@ -23,6 +23,7 @@
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.logging.Level;
 import java.util.logging.Logger;
 import java.util.stream.Collectors;
@@ -43,6 +44,7 @@ public final class JaegerThriftSpanExporter implements SpanExporter {
 
   private final ThrottlingLogger logger =
       new ThrottlingLogger(Logger.getLogger(JaegerThriftSpanExporter.class.getName()));
+  private final AtomicBoolean isShutdown = new AtomicBoolean();
   private final ThriftSender thriftSender;
   private final Process process;
 
@@ -82,13 +84,16 @@ public final class JaegerThriftSpanExporter implements SpanExporter {
    */
   @Override
   public CompletableResultCode export(Collection<SpanData> spans) {
+    if (isShutdown.get()) {
+      return CompletableResultCode.ofFailure();
+    }
+
     Map<Process, List<Span>> batches =
         spans.stream().collect(Collectors.groupingBy(SpanData::getResource)).entrySet().stream()
             .collect(
                 Collectors.toMap(
                     entry -> createProcess(entry.getKey()),
                     entry -> Adapter.toJaeger(entry.getValue())));
-
     List<CompletableResultCode> batchResults = new ArrayList<>(batches.size());
     batches.forEach(
         (process, jaegerSpans) -> {
@@ -148,13 +153,9 @@ public static JaegerThriftSpanExporterBuilder builder() {
    */
   @Override
   public CompletableResultCode shutdown() {
-    CompletableResultCode result = new CompletableResultCode();
-    // todo
-    return result.succeed();
-  }
-
-  // Visible for testing
-  Process getProcess() {
-    return process;
+    if (!isShutdown.compareAndSet(false, true)) {
+      logger.log(Level.WARNING, "Calling shutdown() multiple times.");
+    }
+    return CompletableResultCode.ofSuccess();
   }
 }