More predictable signaling

open-telemetry · Mar 19, 2021 · 37d2863 · 37d2863
1 parent 5ac7737
commit 37d2863
Showing 1 changed file with 5 additions and 4 deletions.
diff --git a/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java b/sdk/trace/src/main/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessor.java
@@ -23,6 +23,7 @@
 import java.util.concurrent.BlockingQueue;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.logging.Level;
 import java.util.logging.Logger;
@@ -131,7 +132,7 @@ private static final class Worker implements Runnable {
 
     private long nextExportTime;
     private final BlockingQueue<ReadableSpan> queue;
-    private final AtomicBoolean needSignal = new AtomicBoolean(false);
+    private final AtomicInteger spansNeeded = new AtomicInteger(Integer.MAX_VALUE);
     private final BlockingQueue<Boolean> signal;
     private final AtomicReference<CompletableResultCode> flushRequested = new AtomicReference<>();
     private volatile boolean continueWork = true;
@@ -182,7 +183,7 @@ private void addSpan(ReadableSpan span) {
       if (!queue.offer(span)) {
         droppedSpans.add(1);
       } else {
-        if (queue.size() >= maxExportBatchSize && needSignal.get()) {
+        if (queue.size() >= spansNeeded.get()) {
           signal.offer(true);
         }
       }
@@ -207,9 +208,9 @@ public void run() {
           try {
             long pollWaitTime = nextExportTime - System.nanoTime();
             if (pollWaitTime > 0) {
-              needSignal.set(true);
+              spansNeeded.set(maxExportBatchSize - batch.size());
               signal.poll(pollWaitTime, TimeUnit.NANOSECONDS);
-              needSignal.set(false);
+              spansNeeded.set(Integer.MAX_VALUE);
             }
           } catch (InterruptedException e) {
             Thread.currentThread().interrupt();