Fix missed wakeup race condition in BatchProcessor

.changelog/5409.fixed

@@ -0,0 +1 @@
+Fix missed wakeup race in `BatchProcessor` that could delay span/log export by a full schedule delay when a batch filled while the worker was clearing its wakeup event

opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py

@@ -154,14 +154,19 @@ def worker(self):
             # https://github.com/open-telemetry/opentelemetry-specification/blob/main/specification/trace/sdk.md#batching-processor.
             # Shutdown will interrupt this sleep. Emit will interrupt this sleep only if the queue is bigger then threshold.
             sleep_interrupted = self._worker_awaken.wait(self._schedule_delay)
+            # Clear the event before exporting, not after. A wakeup signaled by
+            # `emit` between the export loop finishing and a late `clear` would
+            # be erased, leaving a full batch stuck in the queue for up to
+            # `schedule_delay`. Clearing first means any `set` that arrives
+            # while exporting is kept for the next `wait`.
+            self._worker_awaken.clear()
             if self._shutdown:
                 break
             self._export(
                 BatchExportStrategy.EXPORT_WHILE_BATCH_EXCEEDS_THRESHOLD
                 if sleep_interrupted
                 else BatchExportStrategy.EXPORT_AT_LEAST_ONE_BATCH
             )
-            self._worker_awaken.clear()
         self._export(BatchExportStrategy.EXPORT_ALL)
 
     def _export(self, batch_strategy: BatchExportStrategy) -> None:

opentelemetry-sdk/tests/shared_internal/test_batch_processor.py

@@ -119,6 +119,56 @@ def test_telemetry_exported_once_schedule_delay_reached(
         exporter.export.assert_called_once_with([telemetry])
         batch_processor.shutdown()
 
+    # pylint: disable=no-self-use
+    def test_wakeup_not_lost_when_batch_fills_during_event_clear(
+        self, batch_processor_class, telemetry
+    ):
+        # Regression test for a missed wakeup race: if `emit` filled a batch
+        # and set `_worker_awaken` after the worker's export loop drained the
+        # queue but before the worker cleared the event, the signal was erased
+        # and the batch sat in the queue for a full schedule delay.
+        exporter = Mock()
+        batch_processor = batch_processor_class(
+            exporter,
+            max_queue_size=100,
+            max_export_batch_size=10,
+            # Will not be reached during the test; if the wakeup is lost the
+            # asserts below fail long before this delay elapses.
+            schedule_delay_millis=30000,
+            export_timeout_millis=500,
+        )
+        processor = batch_processor._batch_processor
+        original_clear = processor._worker_awaken.clear
+        raced = threading.Event()
+
+        def emit_batch_then_clear():
+            # Fill a full batch (which calls `_worker_awaken.set()`) at the
+            # exact point the worker is about to clear the event, simulating
+            # an application thread that raced with the worker.
+            if not raced.is_set():
+                raced.set()
+                for _ in range(10):
+                    processor.emit(telemetry)
+            original_clear()
+
+        processor._worker_awaken.clear = emit_batch_then_clear
+        # Trigger the first export; the worker wakes, exports this batch, and
+        # then races with the batch emitted from the clear hook.
+        for _ in range(10):
+            processor.emit(telemetry)
+
+        deadline = time.time() + 5
+        while time.time() < deadline:
+            exported = sum(
+                len(call.args[0]) for call in exporter.export.call_args_list
+            )
+            if exported == 20:
+                break
+            time.sleep(0.01)
+        assert exported == 20
+        assert len(processor._queue) == 0
+        batch_processor.shutdown()
+
     def test_telemetry_flushed_before_shutdown_and_dropped_after_shutdown(
         self, batch_processor_class, telemetry
     ):