Make sure forceFlush / shutdown can have callers wait for them to be done by returning CompletableResultCode.

exporters/inmemory/src/main/java/io/opentelemetry/exporters/inmemory/InMemoryMetricExporter.java

@@ -16,7 +16,7 @@
 
 package io.opentelemetry.exporters.inmemory;
 
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.util.ArrayList;

exporters/inmemory/src/main/java/io/opentelemetry/exporters/inmemory/InMemorySpanExporter.java

@@ -16,7 +16,7 @@
 
 package io.opentelemetry.exporters.inmemory;
 
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import java.util.ArrayList;
@@ -108,11 +108,12 @@ public CompletableResultCode flush() {
   }
 
   @Override
-  public void shutdown() {
+  public CompletableResultCode shutdown() {
     synchronized (this) {
       finishedSpanItems.clear();
       isStopped = true;
     }
+    return CompletableResultCode.ofSuccess();
   }
 
   private InMemorySpanExporter() {}

exporters/jaeger/src/main/java/io/opentelemetry/exporters/jaeger/JaegerGrpcSpanExporter.java

@@ -19,13 +19,14 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.MoreExecutors;
+import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.opentelemetry.exporters.jaeger.proto.api_v2.Collector;
 import io.opentelemetry.exporters.jaeger.proto.api_v2.Collector.PostSpansResponse;
 import io.opentelemetry.exporters.jaeger.proto.api_v2.CollectorServiceGrpc;
 import io.opentelemetry.exporters.jaeger.proto.api_v2.Model;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.ConfigBuilder;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
@@ -168,15 +169,21 @@ public static Builder newBuilder() {
 
   /**
    * Initiates an orderly shutdown in which preexisting calls continue but new calls are immediately
-   * cancelled. The channel is forcefully closed after a timeout.
+   * cancelled.
    */
   @Override
-  public void shutdown() {
-    try {
-      managedChannel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
-    } catch (InterruptedException e) {
-      logger.log(Level.WARNING, "Failed to shutdown the gRPC channel", e);
-    }
+  public CompletableResultCode shutdown() {
+    final CompletableResultCode result = new CompletableResultCode();
+    managedChannel.notifyWhenStateChanged(
+        ConnectivityState.SHUTDOWN,
+        new Runnable() {
+          @Override
+          public void run() {
+            result.succeed();
+          }
+        });
+    managedChannel.shutdown();
+    return result;
   }
 
   /** Builder utility for this exporter. */

exporters/logging/src/main/java/io/opentelemetry/exporters/logging/LoggingMetricExporter.java

@@ -16,7 +16,7 @@
 
 package io.opentelemetry.exporters.logging;
 
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;
 import java.util.Collection;

exporters/logging/src/main/java/io/opentelemetry/exporters/logging/LoggingSpanExporter.java

@@ -16,7 +16,7 @@
 
 package io.opentelemetry.exporters.logging;
 
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
 import java.util.Collection;
@@ -55,7 +55,7 @@ public CompletableResultCode flush() {
   }
 
   @Override
-  public void shutdown() {
-    this.flush();
+  public CompletableResultCode shutdown() {
+    return flush();
   }
 }

exporters/logging/src/test/java/io/opentelemetry/exporters/logging/LoggingSpanExporterTest.java

@@ -21,7 +21,7 @@
 
 import io.opentelemetry.common.AttributeValue;
 import io.opentelemetry.common.Attributes;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.TestSpanData;
 import io.opentelemetry.sdk.trace.data.EventImpl;
 import io.opentelemetry.sdk.trace.data.SpanData;

exporters/otlp/src/main/java/io/opentelemetry/exporters/otlp/OtlpGrpcMetricExporter.java

@@ -30,7 +30,7 @@
 import io.opentelemetry.proto.collector.metrics.v1.ExportMetricsServiceResponse;
 import io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc;
 import io.opentelemetry.proto.collector.metrics.v1.MetricsServiceGrpc.MetricsServiceFutureStub;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.ConfigBuilder;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.export.MetricExporter;

exporters/otlp/src/main/java/io/opentelemetry/exporters/otlp/OtlpGrpcSpanExporter.java

@@ -22,6 +22,7 @@
 import com.google.common.util.concurrent.FutureCallback;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.MoreExecutors;
+import io.grpc.ConnectivityState;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Metadata;
@@ -30,7 +31,7 @@
 import io.opentelemetry.proto.collector.trace.v1.ExportTraceServiceResponse;
 import io.opentelemetry.proto.collector.trace.v1.TraceServiceGrpc;
 import io.opentelemetry.proto.collector.trace.v1.TraceServiceGrpc.TraceServiceFutureStub;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.ConfigBuilder;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
@@ -169,15 +170,21 @@ public static OtlpGrpcSpanExporter getDefault() {
 
   /**
    * Initiates an orderly shutdown in which preexisting calls continue but new calls are immediately
-   * cancelled. The channel is forcefully closed after a timeout.
+   * cancelled.
    */
   @Override
-  public void shutdown() {
-    try {
-      managedChannel.shutdown().awaitTermination(5, TimeUnit.SECONDS);
-    } catch (InterruptedException e) {
-      logger.log(Level.WARNING, "Failed to shutdown the gRPC channel", e);
-    }
+  public CompletableResultCode shutdown() {
+    final CompletableResultCode result = new CompletableResultCode();
+    managedChannel.notifyWhenStateChanged(
+        ConnectivityState.SHUTDOWN,
+        new Runnable() {
+          @Override
+          public void run() {
+            result.succeed();
+          }
+        });
+    managedChannel.shutdown();
+    return result;
   }
 
   /** Builder utility for this exporter. */

exporters/zipkin/src/main/java/io/opentelemetry/exporters/zipkin/ZipkinSpanExporter.java

@@ -21,8 +21,8 @@
 import io.opentelemetry.common.AttributeValue;
 import io.opentelemetry.common.ReadableAttributes;
 import io.opentelemetry.common.ReadableKeyValuePairs.KeyValueConsumer;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.ConfigBuilder;
 import io.opentelemetry.sdk.resources.ResourceAttributes;
 import io.opentelemetry.sdk.trace.data.SpanData;
@@ -289,12 +289,13 @@ public CompletableResultCode flush() {
   }
 
   @Override
-  public void shutdown() {
+  public CompletableResultCode shutdown() {
     try {
       sender.close();
     } catch (IOException e) {
       logger.log(Level.WARNING, "Exception while closing the Zipkin Sender instance", e);
     }
+    return CompletableResultCode.ofSuccess();
   }
 
   /**

exporters/zipkin/src/test/java/io/opentelemetry/exporters/zipkin/ZipkinSpanExporterEndToEndHttpTest.java

@@ -20,7 +20,7 @@
 
 import com.google.common.collect.ImmutableList;
 import io.opentelemetry.common.Attributes;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.TestSpanData;
 import io.opentelemetry.sdk.trace.data.EventImpl;
 import io.opentelemetry.sdk.trace.data.SpanData;

exporters/zipkin/src/test/java/io/opentelemetry/exporters/zipkin/ZipkinSpanExporterTest.java

@@ -26,8 +26,8 @@
 import com.google.common.collect.ImmutableList;
 import io.opentelemetry.common.AttributeValue;
 import io.opentelemetry.common.Attributes;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.common.InstrumentationLibraryInfo;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
 import io.opentelemetry.sdk.common.export.ConfigBuilder;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.resources.ResourceAttributes;

sdk/all/src/jmh/java/io/opentelemetry/sdk/trace/SpanPipelineBenchmark.java

@@ -21,7 +21,7 @@
 
 import io.opentelemetry.common.Attributes;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.trace.data.SpanData;
 import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor;
 import io.opentelemetry.sdk.trace.export.SpanExporter;
@@ -98,8 +98,9 @@ public CompletableResultCode flush() {
     }
 
     @Override
-    public void shutdown() {
+    public CompletableResultCode shutdown() {
       // no-op
+      return CompletableResultCode.ofSuccess();
     }
   }
 

sdk/all/src/jmh/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorBenchmark.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2020, OpenTelemetry Authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package io.opentelemetry.sdk.trace.export;
+
+import com.google.common.collect.ImmutableList;
+import io.opentelemetry.OpenTelemetry;
+import io.opentelemetry.sdk.common.CompletableResultCode;
+import io.opentelemetry.sdk.trace.ReadableSpan;
+import io.opentelemetry.sdk.trace.data.SpanData;
+import io.opentelemetry.trace.Span;
+import io.opentelemetry.trace.Tracer;
+import java.util.Collection;
+import java.util.List;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import org.openjdk.jmh.annotations.Benchmark;
+import org.openjdk.jmh.annotations.Fork;
+import org.openjdk.jmh.annotations.Level;
+import org.openjdk.jmh.annotations.Measurement;
+import org.openjdk.jmh.annotations.OutputTimeUnit;
+import org.openjdk.jmh.annotations.Param;
+import org.openjdk.jmh.annotations.Scope;
+import org.openjdk.jmh.annotations.Setup;
+import org.openjdk.jmh.annotations.State;
+import org.openjdk.jmh.annotations.Threads;
+import org.openjdk.jmh.annotations.Warmup;
+
+@State(Scope.Benchmark)
+public class BatchSpanProcessorBenchmark {
+
+  private static class DelayingSpanExporter implements SpanExporter {
+
+    private final ScheduledExecutorService executor;
+
+    private final int delayMs;
+
+    private DelayingSpanExporter(int delayMs) {
+      executor = Executors.newScheduledThreadPool(5);
+      this.delayMs = delayMs;
+    }
+
+    @SuppressWarnings("FutureReturnValueIgnored")
+    @Override
+    public CompletableResultCode export(Collection<SpanData> spans) {
+      final CompletableResultCode result = new CompletableResultCode();
+      executor.schedule(
+          new Runnable() {
+            @Override
+            public void run() {
+              result.succeed();
+            }
+          },
+          delayMs,
+          TimeUnit.MILLISECONDS);
+      return result;
+    }
+
+    @Override
+    public CompletableResultCode flush() {
+      return CompletableResultCode.ofSuccess();
+    }
+
+    @Override
+    public CompletableResultCode shutdown() {
+      return CompletableResultCode.ofSuccess();
+    }
+  }
+
+  @Param({"0", "1", "5"})
+  private int delayMs;
+
+  @Param({"1000", "2000", "5000"})
+  private int spanCount;
+
+  private List<Span> spans;
+
+  private BatchSpanProcessor processor;
+
+  @Setup(Level.Trial)
+  public final void setup() {
+    SpanExporter exporter = new DelayingSpanExporter(delayMs);
+    processor = BatchSpanProcessor.newBuilder(exporter).build();
+
+    ImmutableList.Builder<Span> spans = ImmutableList.builderWithExpectedSize(spanCount);
+    Tracer tracer = OpenTelemetry.getTracerProvider().get("benchmarkTracer");
+    for (int i = 0; i < spanCount; i++) {
+      spans.add(tracer.spanBuilder("span").startSpan());
+    }
+    this.spans = spans.build();
+  }
+
+  /** Export spans through {@link io.opentelemetry.sdk.trace.export.BatchSpanProcessor}. */
+  @Benchmark
+  @Fork(1)
+  @Threads(5)
+  @Warmup(iterations = 5, time = 1)
+  @Measurement(iterations = 10, time = 1)
+  @OutputTimeUnit(TimeUnit.SECONDS)
+  public void export() {
+    for (Span span : spans) {
+      processor.onEnd((ReadableSpan) span);
+    }
+    processor.forceFlush().join(10, TimeUnit.MINUTES);
+  }
+}

sdk/all/src/jmh/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorDroppedSpansBenchmark.java

@@ -18,7 +18,7 @@
 
 import io.opentelemetry.OpenTelemetry;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
-import io.opentelemetry.sdk.common.export.CompletableResultCode;
+import io.opentelemetry.sdk.common.CompletableResultCode;
 import io.opentelemetry.sdk.metrics.data.MetricData;
 import io.opentelemetry.sdk.metrics.data.MetricData.LongPoint;
 import io.opentelemetry.sdk.metrics.data.MetricData.Point;
@@ -58,7 +58,9 @@ public CompletableResultCode flush() {
     }
 
     @Override
-    public void shutdown() {}
+    public CompletableResultCode shutdown() {
+      return CompletableResultCode.ofSuccess();
+    }
   }
 
   @State(Scope.Benchmark)