Actually flush in ExecutorServiceSpanProcessor (#3806)

* Actually flush in ExecutorServiceSpanProcessor

* Fix mock reset

* Remove drift comment

Author: anuraaga

sdk-extensions/tracing-incubator/src/main/java/io/opentelemetry/sdk/extension/incubator/trace/ExecutorServiceSpanProcessor.java

@@ -145,6 +145,7 @@ List<SpanData> getBatch() {
 
   private static class Worker implements Runnable {
 
+    private final AtomicBoolean running = new AtomicBoolean();
     private final AtomicLong nextExportTime = new AtomicLong();
     private final ArrayBlockingQueue<SpanData> batch;
     private final AtomicBoolean isShutdown;
@@ -211,7 +212,9 @@ private void updateNextExportTime() {
 
     @Override
     public void run() {
-      // nextExportTime is set for the first time in the constructor
+      if (!running.compareAndSet(false, true)) {
+        return;
+      }
 
       boolean continueWork = true;
       while (continueWork && !isShutdown.get()) {
@@ -238,6 +241,7 @@ public void run() {
       if (flushRequested.get() != null) {
         workerExporter.flush(batch, queue);
       }
+      running.set(false);
     }
 
     private void scheduleNextRun() {
@@ -275,7 +279,10 @@ public void addSpan(ReadableSpan span) {
     public CompletableResultCode forceFlush() {
       CompletableResultCode flushResult = new CompletableResultCode();
       // we set the atomic here to trigger the worker loop to do a flush on its next iteration.
-      flushRequested.compareAndSet(null, flushResult);
+      if (flushRequested.compareAndSet(null, flushResult)) {
+        executorService.execute(this);
+        return flushResult;
+      }
       CompletableResultCode possibleResult = flushRequested.get();
       // there's a race here where the flush happening in the worker loop could complete before we
       // get what's in the atomic. In that case, just return success, since we know it succeeded in

sdk-extensions/tracing-incubator/src/test/java/io/opentelemetry/sdk/extension/incubator/trace/ExecutorServiceSpanProcessorTest.java

@@ -396,6 +396,7 @@ public void continuesIfExporterTimesOut() throws InterruptedException {
     // Still processing new spans.
     CountDownLatch exportedAgain = new CountDownLatch(1);
     reset(mockSpanExporter);
+    when(mockSpanExporter.shutdown()).thenReturn(CompletableResultCode.ofSuccess());
     when(mockSpanExporter.export(
             argThat(
                 spans -> {
@@ -472,27 +473,27 @@ void exportNotSampledSpans_recordOnly() {
   }
 
   @Test
-  @Timeout(10)
   void shutdownFlushes() {
     WaitingSpanExporter waitingSpanExporter =
         new WaitingSpanExporter(1, CompletableResultCode.ofSuccess());
     // Set the export delay to large value, in order to confirm the #flush() below works
 
-    sdkTracerProvider =
+    SdkTracerProvider sdkTracerProvider =
         SdkTracerProvider.builder()
             .addSpanProcessor(
                 ExecutorServiceSpanProcessor.builder(waitingSpanExporter, executor, false)
                     .setScheduleDelay(10, TimeUnit.SECONDS)
                     .build())
             .build();
 
-    ReadableSpan span2 = createEndedSpan(SPAN_NAME_2);
+    Span span2 = sdkTracerProvider.get("test").spanBuilder(SPAN_NAME_2).startSpan();
+    span2.end();
 
     // Force a shutdown, which forces processing of all remaining spans.
     sdkTracerProvider.shutdown().join(10, TimeUnit.SECONDS);
 
     List<SpanData> exported = waitingSpanExporter.getExported();
-    assertThat(exported).containsExactly(span2.toSpanData());
+    assertThat(exported).containsExactly(((ReadableSpan) span2).toSpanData());
     assertThat(waitingSpanExporter.shutDownCalled.get()).isTrue();
   }