[Bug]: BatchProcessor missed wakeup race condition causes export latency spikes

[pranaysb]

Describe your environment

OS: All
Python version: All (tested 3.12, 3.13)
SDK version: main / 1.43.0
API version: main / 1.43.0

What happened?

There is a severe threading bug in the core BatchProcessor (opentelemetry.sdk._shared_internal.BatchProcessor) that causes span and log exports to stall.
The worker thread uses a threading.Event to wake up when emit() fills the queue. However, the event is cleared after the queue has been processed, but before going to sleep.
If an application emits telemetry and hits max_export_batch_size immediately after _should_export_batch() returns False but before clear() is called in the worker loop, the emit() thread's _worker_awaken.set() signal is blindly erased by the worker. The worker then sleeps for the full schedule_delay (default 5000ms) while a full batch of items sits stuck in the queue.

Steps to Reproduce

The race condition is extremely tight, but can be reproduced deterministically by hooking into the _worker_awaken.clear method on a test BatchProcessor:

import threading, time
from opentelemetry.sdk._shared_internal import BatchProcessor

class DummyExporter:
    def __init__(self): self.exported = []
    def export(self, batch): self.exported.extend(batch)
    def shutdown(self): pass

class DummyMetrics:
    def register_queue_size(self, cb): pass
    def drop_items(self, count): pass
    def finish_items(self, count, error): pass

exporter = DummyExporter()
processor = BatchProcessor(
    exporter, schedule_delay_millis=5000, max_export_batch_size=10, 
    export_timeout_millis=30000, max_queue_size=100, exporting="Test", metrics=DummyMetrics()
)

# Simulate the race condition by hooking into clear()
original_clear = processor._worker_awaken.clear
def hooked_clear():
    # Exactly when the worker is about to clear the event, 
    # we simulate the application thread filling a new batch!
    for i in range(10): processor.emit(i)
    # The worker blindly clears the signal we just generated!
    original_clear()
    
processor._worker_awaken.clear = hooked_clear

# Trigger the first export
for i in range(10): processor.emit(i)
time.sleep(0.5)

print(f"Total exported: {len(exporter.exported)}")
print(f"Items stuck in queue: {len(processor._queue)}")
processor.shutdown()

Expected Result

The worker should safely process wakeups without losing signals. In the reproducer above, the expected result is:
Total exported: 20
Items stuck in queue: 0

Actual Result

The second batch's wakeup signal is erased by the race condition, causing it to sit in the queue for a full 5 seconds (until the wait timeout finishes).
Total exported: 10
Items stuck in queue: 10

Additional context

High-throughput telemetry causes severe batch export delays due to this race condition. Spans and logs can sit in the queue for a full schedule_delay even though the batch size is met, resulting in jagged 5-second latency spikes in high-throughput environments.

Would you like to implement a fix?

Yes

Tip

_{React with 👍 to help prioritize this issue. Please use comments to provide useful context, avoiding +1 or me too, to help us triage it. Learn more here.}