From 0f105cc7fa932dc9819f69d8e73582a27528a296 Mon Sep 17 00:00:00 2001
From: Gautham Prabhu <71874814+GauthamPrabhuM@users.noreply.github.com>
Date: Mon, 13 Jul 2026 12:40:39 +0530
Subject: [PATCH] Fix missed wakeup race condition in BatchProcessor

The worker thread cleared _worker_awaken only after its export loop
finished. If an application thread filled a batch and set the event in
the window between the export loop draining the queue and the worker
calling clear(), the wakeup was erased and the full batch sat in the
queue for up to schedule_delay (5s by default).

Clear the event immediately after wait() returns instead, so any set()
that arrives while the worker is exporting is preserved for the next
wait(). This affects both BatchSpanProcessor and BatchLogRecordProcessor
since they share this implementation.

Fixes #5400

Assisted-by: Claude Fable 5
---
 .changelog/5409.fixed                         |  1 +
 .../sdk/_shared_internal/__init__.py          |  7 ++-
 .../shared_internal/test_batch_processor.py   | 50 +++++++++++++++++++
 3 files changed, 57 insertions(+), 1 deletion(-)
 create mode 100644 .changelog/5409.fixed

diff --git a/.changelog/5409.fixed b/.changelog/5409.fixed
new file mode 100644
index 00000000000..e6ea86159ea
--- /dev/null
+++ b/.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
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py
index 3e2b8a263a4..cefe41aae20 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/_shared_internal/__init__.py
@@ -154,6 +154,12 @@ 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(
@@ -161,7 +167,6 @@ def worker(self):
                 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:
diff --git a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py b/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py
index 02337757073..c0648b5d5c6 100644
--- a/opentelemetry-sdk/tests/shared_internal/test_batch_processor.py
+++ b/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
     ):