From 7885fbb0c55527129274a5dca44b3d78d729d554 Mon Sep 17 00:00:00 2001 From: Ferdinand-MZ Date: Sat, 14 Nov 2026 10:19:47 +0700 Subject: [PATCH 1/2] Fix batch processor flush and shutdown races --- .../logs/export/BatchLogRecordProcessor.java | 134 ++++++++---- .../export/BatchLogRecordProcessorTest.java | 197 +++++++++++++++++- .../sdk/trace/export/BatchSpanProcessor.java | 131 ++++++++---- .../trace/export/BatchSpanProcessorTest.java | 176 +++++++++++++++- 4 files changed, 529 insertions(+), 109 deletions(-) diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java index 13ce175e6b7..b6c12257411 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java @@ -49,6 +49,7 @@ public final class BatchLogRecordProcessor implements LogRecordProcessor { BatchLogRecordProcessor.class.getSimpleName() + "_WorkerThread"; private final Worker worker; + private final Thread workerThread; /** * Returns a new Builder for {@link BatchLogRecordProcessor}. @@ -79,8 +80,8 @@ public static BatchLogRecordProcessorBuilder builder(LogRecordExporter logRecord exporterTimeoutNanos, new ArrayBlockingQueue<>(maxQueueSize), maxQueueSize); // TODO: use JcTools.newFixedSizeQueue(..) - Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); - workerThread.start(); + this.workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); + this.workerThread.start(); } @Override @@ -115,6 +116,11 @@ List getBatch() { return worker.batch; } + // Visible for testing + Thread getWorkerThread() { + return workerThread; + } + @Override public String toString() { return "BatchLogRecordProcessor{" @@ -199,51 +205,78 @@ private void addLog(ReadWriteLogRecord logData) { @Override public void run() { - updateNextExportTime(); + boolean normalExit = false; + try { + updateNextExportTime(); - while (continueWork) { - if (flushRequested.get() != null) { - flush(); - } - while (!queue.isEmpty() && batch.size() < maxExportBatchSize) { - batch.add(queue.poll().toLogRecordData()); + while (continueWork) { + while (continueWork && flushRequested.get() != null) { + flush(); + } + if (!continueWork) { + break; + } + while (!queue.isEmpty() && batch.size() < maxExportBatchSize) { + batch.add(queue.poll().toLogRecordData()); + } + if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { + exportCurrentBatch(); + updateNextExportTime(); + } + if (queue.isEmpty()) { + try { + long pollWaitTime = nextExportTime - System.nanoTime(); + if (pollWaitTime > 0) { + logsNeeded.set(maxExportBatchSize - batch.size()); + signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); + logsNeeded.set(Integer.MAX_VALUE); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } } - if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { - exportCurrentBatch(); - updateNextExportTime(); + if (!continueWork) { + normalExit = true; } - if (queue.isEmpty()) { - try { - long pollWaitTime = nextExportTime - System.nanoTime(); - if (pollWaitTime > 0) { - logsNeeded.set(maxExportBatchSize - batch.size()); - signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); - logsNeeded.set(Integer.MAX_VALUE); - } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; + } finally { + continueWork = false; + CompletableResultCode sentinel = + normalExit ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); + CompletableResultCode flushResult = flushRequested.getAndSet(sentinel); + if (flushResult != null) { + if (normalExit) { + flushResult.succeed(); + } else { + flushResult.fail(); } } } } private void flush() { - int logsToFlush = queue.size(); - while (logsToFlush > 0) { - ReadWriteLogRecord logRecord = queue.poll(); - assert logRecord != null; - batch.add(logRecord.toLogRecordData()); - logsToFlush--; - if (batch.size() >= maxExportBatchSize) { - exportCurrentBatch(); + CompletableResultCode flushResult = flushRequested.getAndSet(null); + try { + int logsToFlush = queue.size(); + while (logsToFlush > 0) { + ReadWriteLogRecord logRecord = queue.poll(); + assert logRecord != null; + batch.add(logRecord.toLogRecordData()); + logsToFlush--; + if (batch.size() >= maxExportBatchSize) { + exportCurrentBatch(); + } } - } - exportCurrentBatch(); - CompletableResultCode flushResult = flushRequested.get(); - if (flushResult != null) { - flushResult.succeed(); - flushRequested.set(null); + exportCurrentBatch(); + if (flushResult != null) { + flushResult.succeed(); + } + } catch (Throwable t) { + if (flushResult != null) { + flushResult.fail(); + } + throw t; } } @@ -274,16 +307,25 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = new CompletableResultCode(); - // we set the atomic here to trigger the worker loop to do a flush of the entire queue. - if (flushRequested.compareAndSet(null, flushResult)) { - signal.offer(true); + CompletableResultCode flushResult = null; + while (true) { + CompletableResultCode current = flushRequested.get(); + if (current != null) { + return current; + } + if (!continueWork) { + return isShutdown.get() + ? CompletableResultCode.ofSuccess() + : CompletableResultCode.ofFailure(); + } + if (flushResult == null) { + flushResult = new CompletableResultCode(); + } + if (flushRequested.compareAndSet(null, flushResult)) { + signal.offer(true); + 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 - // the interim. - return possibleResult == null ? CompletableResultCode.ofSuccess() : possibleResult; } private void exportCurrentBatch() { diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java index 1b86ba63dee..a93695e20f3 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java @@ -29,10 +29,12 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicInteger; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -511,6 +513,168 @@ void toString_Valid() { + "}"); } + @Test + void shutdown_whileForceFlushInFlight_exportsLogsEmittedBeforeShutdown() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + // Wait until the worker enters exportCurrentBatch() for LOG_MESSAGE_1 during flush(). + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + // Log 2 is emitted while the first forceFlush() is still in flight, before shutdown() is + // called. + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + + // Trigger shutdown while the first forceFlush() is in flight. + CompletableResultCode shutdownResult = processor.shutdown(); + + // Unblock the in-flight export. + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2)); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileForceFlushInFlight_queuesBehindIt() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + CompletableResultCode secondFlush = processor.forceFlush(); + CompletableResultCode thirdFlush = processor.forceFlush(); + + // Complete only the first export; secondFlush and thirdFlush must wait for the second export. + exporter.completeExport(0); + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2))); + assertThat(secondFlush.isDone()).isFalse(); + assertThat(thirdFlush.isDone()).isFalse(); + + exporter.succeed(); + secondFlush.join(10, TimeUnit.SECONDS); + thirdFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(thirdFlush.isSuccess()).isTrue(); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_fromWhenCompleteCallback_exportsNewlyQueuedLogs() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode firstFlush = processor.forceFlush(); + AtomicReference secondFlushRef = new AtomicReference<>(); + + firstFlush.whenComplete(() -> { + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + secondFlushRef.set(processor.forceFlush()); + }); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + CompletableResultCode secondFlush = secondFlushRef.get(); + assertThat(secondFlush).isNotNull(); + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .satisfiesExactly( + log -> assertThat(log).hasBody(LOG_MESSAGE_1), + log -> assertThat(log).hasBody(LOG_MESSAGE_2)); + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_afterShutdown_completes() { + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter).build(); + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + CompletableResultCode secondShutdownResult = processor.shutdown(); + secondShutdownResult.join(1, TimeUnit.SECONDS); + assertThat(secondShutdownResult.isDone()).isTrue(); + assertThat(secondShutdownResult.isSuccess()).isTrue(); + + CompletableResultCode flushAfterShutdown = processor.forceFlush(); + flushAfterShutdown.join(1, TimeUnit.SECONDS); + assertThat(flushAfterShutdown.isDone()).isTrue(); + assertThat(flushAfterShutdown.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws Exception { + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + processor.getWorkerThread().interrupt(); + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode flushAfterInterruption = processor.forceFlush(); + flushAfterInterruption.join(1, TimeUnit.SECONDS); + assertThat(flushAfterInterruption.isDone()).isTrue(); + assertThat(flushAfterInterruption.isSuccess()).isFalse(); + + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(1, TimeUnit.SECONDS); + assertThat(shutdownResult.isDone()).isTrue(); + assertThat(shutdownResult.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + private static final class BlockingLogRecordExporter implements LogRecordExporter { final Object monitor = new Object(); @@ -574,9 +738,11 @@ private void unblock() { private static class CompletableLogRecordExporter implements LogRecordExporter { - private final List results = new ArrayList<>(); + private final Object lock = new Object(); + + private final List results = new CopyOnWriteArrayList<>(); - private final List exported = new ArrayList<>(); + private final List exported = new CopyOnWriteArrayList<>(); private volatile boolean succeeded; @@ -584,20 +750,31 @@ List getExported() { return exported; } + void completeExport(int index) { + results.get(index).succeed(); + } + void succeed() { - succeeded = true; - results.forEach(CompletableResultCode::succeed); + List toComplete; + synchronized (lock) { + succeeded = true; + toComplete = new ArrayList<>(results); + } + toComplete.forEach(CompletableResultCode::succeed); } @Override public CompletableResultCode export(Collection logs) { - exported.addAll(logs); - if (succeeded) { - return CompletableResultCode.ofSuccess(); + synchronized (lock) { + if (succeeded) { + exported.addAll(logs); + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + results.add(result); + exported.addAll(logs); + return result; } - CompletableResultCode result = new CompletableResultCode(); - results.add(result); - return result; } @Override 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 index 422c9a576c1..7c9dceffd6f 100644 --- 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 @@ -54,6 +54,7 @@ public final class BatchSpanProcessor implements SpanProcessor { private final boolean exportUnsampledSpans; private final Worker worker; + private final Thread workerThread; /** * Returns a new {@link BatchSpanProcessor} with default configuration which batches spans @@ -103,8 +104,8 @@ public static BatchSpanProcessorBuilder builder(SpanExporter spanExporter) { exporterTimeoutNanos, JcTools.newFixedSizeQueue(maxQueueSize), maxQueueSize); - Thread workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); - workerThread.start(); + this.workerThread = new DaemonThreadFactory(WORKER_THREAD_NAME).newThread(worker); + this.workerThread.start(); } @Override @@ -156,6 +157,11 @@ Queue getQueue() { return worker.queue; } + // Visible for testing + Thread getWorkerThread() { + return workerThread; + } + @Override public String toString() { return "BatchSpanProcessor{" @@ -243,29 +249,50 @@ private void addSpan(ReadableSpan span) { @Override public void run() { - updateNextExportTime(); + boolean normalExit = false; + try { + updateNextExportTime(); - while (continueWork) { - if (flushRequested.get() != null) { - flush(); - } - drain(maxExportBatchSize - batch.size()); + while (continueWork) { + while (continueWork && flushRequested.get() != null) { + flush(); + } + if (!continueWork) { + break; + } + drain(maxExportBatchSize - batch.size()); - if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { - exportCurrentBatch(); - updateNextExportTime(); - } - if (queue.isEmpty()) { - try { - long pollWaitTime = nextExportTime - System.nanoTime(); - if (pollWaitTime > 0) { - spansNeeded.set(maxExportBatchSize - batch.size()); - signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); - spansNeeded.set(Integer.MAX_VALUE); + if (batch.size() >= maxExportBatchSize || System.nanoTime() >= nextExportTime) { + exportCurrentBatch(); + updateNextExportTime(); + } + if (queue.isEmpty()) { + try { + long pollWaitTime = nextExportTime - System.nanoTime(); + if (pollWaitTime > 0) { + spansNeeded.set(maxExportBatchSize - batch.size()); + signal.poll(pollWaitTime, TimeUnit.NANOSECONDS); + spansNeeded.set(Integer.MAX_VALUE); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; } - } catch (InterruptedException e) { - Thread.currentThread().interrupt(); - return; + } + } + if (!continueWork) { + normalExit = true; + } + } finally { + continueWork = false; + CompletableResultCode sentinel = + normalExit ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); + CompletableResultCode flushResult = flushRequested.getAndSet(sentinel); + if (flushResult != null) { + if (normalExit) { + flushResult.succeed(); + } else { + flushResult.fail(); } } } @@ -278,19 +305,28 @@ private int drain(int limit) { } private void flush() { - int spansToFlush = queueSize.get(); - while (spansToFlush > 0) { - int drained = drain(maxExportBatchSize - batch.size()); - spansToFlush -= drained; - if (batch.size() >= maxExportBatchSize) { - exportCurrentBatch(); + CompletableResultCode flushResult = flushRequested.getAndSet(null); + try { + int spansToFlush = queueSize.get(); + while (spansToFlush > 0) { + int drained = drain(maxExportBatchSize - batch.size()); + spansToFlush -= drained; + if (drained == 0 && batch.size() < maxExportBatchSize) { + break; + } + if (batch.size() >= maxExportBatchSize) { + exportCurrentBatch(); + } } - } - exportCurrentBatch(); - CompletableResultCode flushResult = flushRequested.get(); - if (flushResult != null) { - flushResult.succeed(); - flushRequested.set(null); + exportCurrentBatch(); + if (flushResult != null) { + flushResult.succeed(); + } + } catch (Throwable t) { + if (flushResult != null) { + flushResult.fail(); + } + throw t; } } @@ -321,16 +357,25 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = new CompletableResultCode(); - // we set the atomic here to trigger the worker loop to do a flush of the entire queue. - if (flushRequested.compareAndSet(null, flushResult)) { - signal.offer(true); + CompletableResultCode flushResult = null; + while (true) { + CompletableResultCode current = flushRequested.get(); + if (current != null) { + return current; + } + if (!continueWork) { + return isShutdown.get() + ? CompletableResultCode.ofSuccess() + : CompletableResultCode.ofFailure(); + } + if (flushResult == null) { + flushResult = new CompletableResultCode(); + } + if (flushRequested.compareAndSet(null, flushResult)) { + signal.offer(true); + 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 - // the interim. - return possibleResult == null ? CompletableResultCode.ofSuccess() : possibleResult; } private void exportCurrentBatch() { diff --git a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java index 2f1b3d005e4..aedcd0b5159 100644 --- a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java +++ b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java @@ -34,9 +34,11 @@ import java.util.Arrays; import java.util.Collection; import java.util.List; +import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicReference; import javax.annotation.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -730,6 +732,147 @@ void exporterThrowsNonRuntimeException() { await().untilAsserted(() -> assertThat(batchSpanProcessor.getQueue()).isEmpty()); } + @Test + void shutdown_whileForceFlushInFlight_exportsSpansEndedBeforeShutdown() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + // Wait until the worker enters exportCurrentBatch() for span1 during flush(). + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + // Span 2 ends while the first forceFlush() is still in flight, before shutdown() is called. + ReadableSpan span2 = createEndedSpan(SPAN_NAME_2); + assertThat(processor.getQueue()).hasSize(1); + + // Trigger shutdown while the first forceFlush() is in flight. + CompletableResultCode shutdownResult = processor.shutdown(); + + // Unblock the in-flight export. + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(exporter.getExported()).containsExactly(span1.toSpanData(), span2.toSpanData()); + } + + @Test + void forceFlush_whileForceFlushInFlight_queuesBehindIt() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + ReadableSpan span2 = createEndedSpan(SPAN_NAME_2); + CompletableResultCode secondFlush = processor.forceFlush(); + CompletableResultCode thirdFlush = processor.forceFlush(); + + // Complete only the first export; secondFlush and thirdFlush must wait for the second export. + exporter.completeExport(0); + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .containsExactly(span1.toSpanData(), span2.toSpanData())); + assertThat(secondFlush.isDone()).isFalse(); + assertThat(thirdFlush.isDone()).isFalse(); + + exporter.succeed(); + secondFlush.join(10, TimeUnit.SECONDS); + thirdFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(thirdFlush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_fromWhenCompleteCallback_exportsNewlyQueuedSpans() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode firstFlush = processor.forceFlush(); + AtomicReference span2Ref = new AtomicReference<>(); + AtomicReference secondFlushRef = new AtomicReference<>(); + + firstFlush.whenComplete(() -> { + span2Ref.set(createEndedSpan(SPAN_NAME_2)); + secondFlushRef.set(processor.forceFlush()); + }); + + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + exporter.succeed(); + + firstFlush.join(10, TimeUnit.SECONDS); + assertThat(firstFlush.isSuccess()).isTrue(); + CompletableResultCode secondFlush = secondFlushRef.get(); + assertThat(secondFlush).isNotNull(); + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isSuccess()).isTrue(); + assertThat(exporter.getExported()) + .containsExactly(span1.toSpanData(), span2Ref.get().toSpanData()); + } + + @Test + void forceFlush_afterShutdown_completes() { + BatchSpanProcessor processor = BatchSpanProcessor.builder(mockSpanExporter).build(); + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + CompletableResultCode secondShutdownResult = processor.shutdown(); + secondShutdownResult.join(1, TimeUnit.SECONDS); + assertThat(secondShutdownResult.isDone()).isTrue(); + assertThat(secondShutdownResult.isSuccess()).isTrue(); + + CompletableResultCode flushAfterShutdown = processor.forceFlush(); + flushAfterShutdown.join(1, TimeUnit.SECONDS); + assertThat(flushAfterShutdown.isDone()).isTrue(); + assertThat(flushAfterShutdown.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws Exception { + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + processor.getWorkerThread().interrupt(); + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode flushAfterInterruption = processor.forceFlush(); + flushAfterInterruption.join(1, TimeUnit.SECONDS); + assertThat(flushAfterInterruption.isDone()).isTrue(); + assertThat(flushAfterInterruption.isSuccess()).isFalse(); + + CompletableResultCode shutdownResult = processor.shutdown(); + shutdownResult.join(1, TimeUnit.SECONDS); + assertThat(shutdownResult.isDone()).isTrue(); + assertThat(shutdownResult.isSuccess()).isFalse(); + } + private static final class BlockingSpanExporter implements SpanExporter { final Object monitor = new Object(); @@ -793,9 +936,11 @@ private void unblock() { private static class CompletableSpanExporter implements SpanExporter { - private final List results = new ArrayList<>(); + private final Object lock = new Object(); + + private final List results = new CopyOnWriteArrayList<>(); - private final List exported = new ArrayList<>(); + private final List exported = new CopyOnWriteArrayList<>(); private volatile boolean succeeded; @@ -803,20 +948,31 @@ List getExported() { return exported; } + void completeExport(int index) { + results.get(index).succeed(); + } + void succeed() { - succeeded = true; - results.forEach(CompletableResultCode::succeed); + List toComplete; + synchronized (lock) { + succeeded = true; + toComplete = new ArrayList<>(results); + } + toComplete.forEach(CompletableResultCode::succeed); } @Override public CompletableResultCode export(Collection spans) { - exported.addAll(spans); - if (succeeded) { - return CompletableResultCode.ofSuccess(); + synchronized (lock) { + if (succeeded) { + exported.addAll(spans); + return CompletableResultCode.ofSuccess(); + } + CompletableResultCode result = new CompletableResultCode(); + results.add(result); + exported.addAll(spans); + return result; } - CompletableResultCode result = new CompletableResultCode(); - results.add(result); - return result; } @Override From dfcfeb8f813ce60f7d8ecc3d6d0854130dfb0c5b Mon Sep 17 00:00:00 2001 From: Ferdinand-MZ Date: Sat, 14 Nov 2026 12:25:58 +0700 Subject: [PATCH 2/2] Improve test coverage and refine batch processor lifecycle handling - Refine worker loop to immediately exit upon thread interruption and complete pending flushes with failure. - Ensure pending flush requests are failed if an unexpected exception occurs during queue drainage in flush(). - Add comprehensive regression tests for concurrent forceFlush, trailing flushes during shutdown, exporter failures, worker interruption, and queue drainage exceptions. - Ensure full deterministic synchronization across all new test cases. --- .../logs/export/BatchLogRecordProcessor.java | 15 +- .../export/BatchLogRecordProcessorTest.java | 245 ++++++++++++++++++ .../sdk/trace/export/BatchSpanProcessor.java | 18 +- .../trace/export/BatchSpanProcessorTest.java | 197 ++++++++++++++ 4 files changed, 456 insertions(+), 19 deletions(-) diff --git a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java index b6c12257411..c996336697b 100644 --- a/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java +++ b/sdk/logs/src/main/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessor.java @@ -209,11 +209,13 @@ public void run() { try { updateNextExportTime(); - while (continueWork) { - while (continueWork && flushRequested.get() != null) { + while (!Thread.currentThread().isInterrupted() && continueWork) { + while (!Thread.currentThread().isInterrupted() + && continueWork + && flushRequested.get() != null) { flush(); } - if (!continueWork) { + if (!continueWork || Thread.currentThread().isInterrupted()) { break; } while (!queue.isEmpty() && batch.size() < maxExportBatchSize) { @@ -237,7 +239,7 @@ public void run() { } } } - if (!continueWork) { + if (!continueWork && !Thread.currentThread().isInterrupted()) { normalExit = true; } } finally { @@ -307,7 +309,6 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = null; while (true) { CompletableResultCode current = flushRequested.get(); if (current != null) { @@ -318,9 +319,7 @@ private CompletableResultCode forceFlush() { ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); } - if (flushResult == null) { - flushResult = new CompletableResultCode(); - } + CompletableResultCode flushResult = new CompletableResultCode(); if (flushRequested.compareAndSet(null, flushResult)) { signal.offer(true); return flushResult; diff --git a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java index a93695e20f3..a77077666c8 100644 --- a/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java +++ b/sdk/logs/src/test/java/io/opentelemetry/sdk/logs/export/BatchLogRecordProcessorTest.java @@ -14,14 +14,17 @@ import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doAnswer; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; import io.opentelemetry.api.internal.GuardedBy; +import io.opentelemetry.context.Context; import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.common.CompletableResultCode; +import io.opentelemetry.sdk.logs.ReadWriteLogRecord; import io.opentelemetry.sdk.logs.SdkLoggerProvider; import io.opentelemetry.sdk.logs.data.LogRecordData; import java.time.Duration; @@ -658,6 +661,7 @@ void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws SdkLoggerProvider sdkLoggerProvider = SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + await().until(() -> processor.getWorkerThread().getState() == Thread.State.TIMED_WAITING); processor.getWorkerThread().interrupt(); processor.getWorkerThread().join(5000); assertThat(processor.getWorkerThread().isAlive()).isFalse(); @@ -675,6 +679,247 @@ void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); } + @Test + @SuppressLogger(BatchLogRecordProcessor.class) + void forceFlush_whenExporterThrows_completesFlushSuccessfully() { + when(mockLogRecordExporter.export(anyList())).thenThrow(new RuntimeException("export failed")); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(processor.getWorkerThread().isAlive()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whenFlushThrows_failsPendingFlushAndTerminatesWorker() throws Exception { + ReadWriteLogRecord mockRecord = mock(ReadWriteLogRecord.class); + when(mockRecord.toLogRecordData()).thenThrow(new RuntimeException("toLogRecordData failed")); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + processor.onEmit(Context.root(), mockRecord); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isFalse(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode subsequentFlush = processor.forceFlush(); + assertThat(subsequentFlush.isDone()).isTrue(); + assertThat(subsequentFlush.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileShutdownExportInFlight_completesPendingFlushOnNormalExit() { + CompletableLogRecordExporter exporter = new CompletableLogRecordExporter(); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until the worker enters exportCurrentBatch() for log1 during shutdown flush. + await() + .untilAsserted(() -> + assertThat(exporter.getExported()) + .satisfiesExactly(log -> assertThat(log).hasBody(LOG_MESSAGE_1))); + + // Queue a trailing forceFlush() while the shutdown export is in flight + CompletableResultCode trailingFlush = processor.forceFlush(); + assertThat(trailingFlush.isDone()).isFalse(); + + // Unblock the export + exporter.succeed(); + + shutdownResult.join(10, TimeUnit.SECONDS); + trailingFlush.join(10, TimeUnit.SECONDS); + + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(trailingFlush.isSuccess()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_whileWorkerInterruptedWithPendingFlush_failsPendingFlush() throws Exception { + CountDownLatch exportStarted = new CountDownLatch(1); + CountDownLatch exportBlocker = new CountDownLatch(1); + when(mockLogRecordExporter.export(anyList())) + .thenAnswer(invocation -> { + exportStarted.countDown(); + try { + exportBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofFailure(); + }); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + processor.forceFlush(); + + exportStarted.await(); + + // Queue a second flush while worker is inside export + CompletableResultCode secondFlush = processor.forceFlush(); + assertThat(secondFlush.isDone()).isFalse(); + + // Interrupt worker thread while second flush is pending + processor.getWorkerThread().interrupt(); + exportBlocker.countDown(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isDone()).isTrue(); + assertThat(secondFlush.isSuccess()).isFalse(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_duringExporterShutdown_returnsSuccess() throws Exception { + CountDownLatch shutdownStarted = new CountDownLatch(1); + CountDownLatch shutdownBlocker = new CountDownLatch(1); + when(mockLogRecordExporter.shutdown()) + .thenAnswer(invocation -> { + shutdownStarted.countDown(); + try { + shutdownBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofSuccess(); + }); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until exporter.shutdown() is in flight + shutdownStarted.await(); + + // While exporter.shutdown() is in flight, continueWork is false and flushRequested is null + CompletableResultCode flushDuringShutdown = processor.forceFlush(); + assertThat(flushDuringShutdown.isDone()).isTrue(); + assertThat(flushDuringShutdown.isSuccess()).isTrue(); + + shutdownBlocker.countDown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_withMultipleBatches_exportsAllBatches() { + WaitingLogRecordExporter exporter = + new WaitingLogRecordExporter(3, CompletableResultCode.ofSuccess()); + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(exporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .setMaxExportBatchSize(2) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + emitLog(sdkLoggerProvider, LOG_MESSAGE_2); + emitLog(sdkLoggerProvider, "log 3"); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(exporter.waitForExport()).hasSize(3); + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + + @Test + void forceFlush_concurrentCalls_allSucceed() throws Exception { + when(mockLogRecordExporter.export(anyList())).thenReturn(CompletableResultCode.ofSuccess()); + + BatchLogRecordProcessor processor = + BatchLogRecordProcessor.builder(mockLogRecordExporter) + .setScheduleDelay(10, TimeUnit.SECONDS) + .build(); + SdkLoggerProvider sdkLoggerProvider = + SdkLoggerProvider.builder().addLogRecordProcessor(processor).build(); + + emitLog(sdkLoggerProvider, LOG_MESSAGE_1); + + int threadCount = 8; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + List results = new CopyOnWriteArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + new Thread( + () -> { + try { + startLatch.await(); + results.add(processor.forceFlush()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }) + .start(); + } + + startLatch.countDown(); + doneLatch.await(5, TimeUnit.SECONDS); + + assertThat(results).hasSize(threadCount); + for (CompletableResultCode result : results) { + result.join(10, TimeUnit.SECONDS); + assertThat(result.isSuccess()).isTrue(); + } + + sdkLoggerProvider.shutdown().join(10, TimeUnit.SECONDS); + } + private static final class BlockingLogRecordExporter implements LogRecordExporter { final Object monitor = new Object(); 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 index 7c9dceffd6f..0aa041e1bf0 100644 --- 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 @@ -253,11 +253,13 @@ public void run() { try { updateNextExportTime(); - while (continueWork) { - while (continueWork && flushRequested.get() != null) { + while (!Thread.currentThread().isInterrupted() && continueWork) { + while (!Thread.currentThread().isInterrupted() + && continueWork + && flushRequested.get() != null) { flush(); } - if (!continueWork) { + if (!continueWork || Thread.currentThread().isInterrupted()) { break; } drain(maxExportBatchSize - batch.size()); @@ -280,7 +282,7 @@ public void run() { } } } - if (!continueWork) { + if (!continueWork && !Thread.currentThread().isInterrupted()) { normalExit = true; } } finally { @@ -311,9 +313,6 @@ private void flush() { while (spansToFlush > 0) { int drained = drain(maxExportBatchSize - batch.size()); spansToFlush -= drained; - if (drained == 0 && batch.size() < maxExportBatchSize) { - break; - } if (batch.size() >= maxExportBatchSize) { exportCurrentBatch(); } @@ -357,7 +356,6 @@ private CompletableResultCode shutdown() { } private CompletableResultCode forceFlush() { - CompletableResultCode flushResult = null; while (true) { CompletableResultCode current = flushRequested.get(); if (current != null) { @@ -368,9 +366,7 @@ private CompletableResultCode forceFlush() { ? CompletableResultCode.ofSuccess() : CompletableResultCode.ofFailure(); } - if (flushResult == null) { - flushResult = new CompletableResultCode(); - } + CompletableResultCode flushResult = new CompletableResultCode(); if (flushRequested.compareAndSet(null, flushResult)) { signal.offer(true); return flushResult; diff --git a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java index aedcd0b5159..fab9bd4ade9 100644 --- a/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java +++ b/sdk/trace/src/test/java/io/opentelemetry/sdk/trace/export/BatchSpanProcessorTest.java @@ -13,6 +13,7 @@ import static org.mockito.ArgumentMatchers.anyList; import static org.mockito.ArgumentMatchers.argThat; import static org.mockito.Mockito.doThrow; +import static org.mockito.Mockito.mock; import static org.mockito.Mockito.reset; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; @@ -21,6 +22,9 @@ import io.github.netmikey.logunit.api.LogCapturer; import io.opentelemetry.api.internal.GuardedBy; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; +import io.opentelemetry.api.trace.TraceFlags; +import io.opentelemetry.api.trace.TraceState; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.internal.testing.slf4j.SuppressLogger; import io.opentelemetry.sdk.common.CompletableResultCode; @@ -858,6 +862,7 @@ void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + await().until(() -> processor.getWorkerThread().getState() == Thread.State.TIMED_WAITING); processor.getWorkerThread().interrupt(); processor.getWorkerThread().join(5000); assertThat(processor.getWorkerThread().isAlive()).isFalse(); @@ -873,6 +878,198 @@ void forceFlush_whenWorkerInterrupted_failsPendingAndSubsequentFlushes() throws assertThat(shutdownResult.isSuccess()).isFalse(); } + @Test + @SuppressLogger(BatchSpanProcessor.class) + void forceFlush_whenExporterThrows_completesFlushSuccessfully() { + when(mockSpanExporter.export(anyList())).thenThrow(new RuntimeException("export failed")); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isTrue(); + assertThat(processor.getWorkerThread().isAlive()).isTrue(); + } + + @Test + void forceFlush_whenFlushThrows_failsPendingFlushAndTerminatesWorker() throws Exception { + ReadableSpan mockSpan = mock(ReadableSpan.class); + SpanContext sampledContext = + SpanContext.create( + "00000000000000000000000000000001", + "0000000000000001", + TraceFlags.getSampled(), + TraceState.getDefault()); + when(mockSpan.getSpanContext()).thenReturn(sampledContext); + when(mockSpan.toSpanData()).thenThrow(new RuntimeException("toSpanData failed")); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + processor.onEnd(mockSpan); + + CompletableResultCode flushResult = processor.forceFlush(); + flushResult.join(10, TimeUnit.SECONDS); + + assertThat(flushResult.isDone()).isTrue(); + assertThat(flushResult.isSuccess()).isFalse(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + CompletableResultCode subsequentFlush = processor.forceFlush(); + assertThat(subsequentFlush.isDone()).isTrue(); + assertThat(subsequentFlush.isSuccess()).isFalse(); + } + + @Test + void forceFlush_whileShutdownExportInFlight_completesPendingFlushOnNormalExit() { + CompletableSpanExporter exporter = new CompletableSpanExporter(); + BatchSpanProcessor processor = + BatchSpanProcessor.builder(exporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + ReadableSpan span1 = createEndedSpan(SPAN_NAME_1); + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until the worker enters exportCurrentBatch() for span1 during shutdown flush. + await() + .untilAsserted( + () -> assertThat(exporter.getExported()).containsExactly(span1.toSpanData())); + + // Queue a trailing forceFlush() while the shutdown export is in flight + CompletableResultCode trailingFlush = processor.forceFlush(); + assertThat(trailingFlush.isDone()).isFalse(); + + // Unblock the export + exporter.succeed(); + + shutdownResult.join(10, TimeUnit.SECONDS); + trailingFlush.join(10, TimeUnit.SECONDS); + + assertThat(shutdownResult.isSuccess()).isTrue(); + assertThat(trailingFlush.isSuccess()).isTrue(); + } + + @Test + void forceFlush_whileWorkerInterruptedWithPendingFlush_failsPendingFlush() throws Exception { + CountDownLatch exportStarted = new CountDownLatch(1); + CountDownLatch exportBlocker = new CountDownLatch(1); + when(mockSpanExporter.export(anyList())) + .thenAnswer(invocation -> { + exportStarted.countDown(); + try { + exportBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofFailure(); + }); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + processor.forceFlush(); + + exportStarted.await(); + + // Queue a second flush while worker is inside export + CompletableResultCode secondFlush = processor.forceFlush(); + assertThat(secondFlush.isDone()).isFalse(); + + // Interrupt worker thread while second flush is pending + processor.getWorkerThread().interrupt(); + exportBlocker.countDown(); + + processor.getWorkerThread().join(5000); + assertThat(processor.getWorkerThread().isAlive()).isFalse(); + + secondFlush.join(10, TimeUnit.SECONDS); + assertThat(secondFlush.isDone()).isTrue(); + assertThat(secondFlush.isSuccess()).isFalse(); + } + + @Test + void forceFlush_duringExporterShutdown_returnsSuccess() throws Exception { + CountDownLatch shutdownStarted = new CountDownLatch(1); + CountDownLatch shutdownBlocker = new CountDownLatch(1); + when(mockSpanExporter.shutdown()) + .thenAnswer(invocation -> { + shutdownStarted.countDown(); + try { + shutdownBlocker.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + return CompletableResultCode.ofSuccess(); + }); + + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + CompletableResultCode shutdownResult = processor.shutdown(); + + // Wait until exporter.shutdown() is in flight + shutdownStarted.await(); + + // While exporter.shutdown() is in flight, continueWork is false and flushRequested is null + CompletableResultCode flushDuringShutdown = processor.forceFlush(); + assertThat(flushDuringShutdown.isDone()).isTrue(); + assertThat(flushDuringShutdown.isSuccess()).isTrue(); + + shutdownBlocker.countDown(); + shutdownResult.join(10, TimeUnit.SECONDS); + assertThat(shutdownResult.isSuccess()).isTrue(); + } + + @Test + void forceFlush_concurrentCalls_allSucceed() throws Exception { + BatchSpanProcessor processor = + BatchSpanProcessor.builder(mockSpanExporter).setScheduleDelay(10, TimeUnit.SECONDS).build(); + sdkTracerProvider = SdkTracerProvider.builder().addSpanProcessor(processor).build(); + + createEndedSpan(SPAN_NAME_1); + + int threadCount = 8; + CountDownLatch startLatch = new CountDownLatch(1); + CountDownLatch doneLatch = new CountDownLatch(threadCount); + List results = new CopyOnWriteArrayList<>(); + + for (int i = 0; i < threadCount; i++) { + new Thread( + () -> { + try { + startLatch.await(); + results.add(processor.forceFlush()); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } finally { + doneLatch.countDown(); + } + }) + .start(); + } + + startLatch.countDown(); + doneLatch.await(5, TimeUnit.SECONDS); + + assertThat(results).hasSize(threadCount); + for (CompletableResultCode result : results) { + result.join(10, TimeUnit.SECONDS); + assertThat(result.isSuccess()).isTrue(); + } + } + private static final class BlockingSpanExporter implements SpanExporter { final Object monitor = new Object();