fix: respect timeout_millis in BatchProcessor.force_flush

Fixes #4568.

Author: chimchim89

CHANGELOG.md

@@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-sdk`: fix `BatchProcessor.force_flush` to respect `timeout_millis`, previously the timeout was ignored and the flush would block until all telemetry was exported
+  ([#4982](https://github.com/open-telemetry/opentelemetry-python/pull/4982))
 - `opentelemetry-sdk`: Add file configuration support with YAML/JSON loading, environment variable substitution, and schema validation against the vendored OTel config JSON schema
   ([#4898](https://github.com/open-telemetry/opentelemetry-python/pull/4898))
 - Fix intermittent CI failures in `getting-started` and `tracecontext` jobs caused by GitHub git CDN SHA propagation lag by installing contrib packages from the already-checked-out local copy instead of a second git clone

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

@@ -18,6 +18,7 @@
 import enum
 import inspect
 import logging
+import math
 import os
 import threading
 import time
@@ -172,18 +173,15 @@ def worker(self):
     def _export(
         self,
         batch_strategy: BatchExportStrategy,
-        flush_should_end: Optional[float] = None,
+        deadline: float = math.inf,
     ) -> bool:
-        # Returns True if all batches were exported, False if flush_should_end was reached.
+        # Returns True if all batches were exported, False if deadline was reached.
         with self._export_lock:
             iteration = 0
             # We could see concurrent export calls from worker and force_flush. We call _should_export_batch
             # once the lock is obtained to see if we still need to make the requested export.
             while self._should_export_batch(batch_strategy, iteration):
-                if (
-                    flush_should_end is not None
-                    and time.time() >= flush_should_end
-                ):
+                if time.time() >= deadline:
                     return False
                 iteration += 1
                 token = attach(set_value(_SUPPRESS_INSTRUMENTATION_KEY, True))
@@ -250,10 +248,10 @@ def shutdown(self, timeout_millis: int = 30000):
     def force_flush(self, timeout_millis: Optional[int] = None) -> bool:
         if self._shutdown:
             return False
-        flush_should_end = (
+        deadline = (
             time.time() + (timeout_millis / 1000)
             if timeout_millis is not None
-            else None
+            else math.inf
         )
         # Blocking call to export.
-        return self._export(BatchExportStrategy.EXPORT_ALL, flush_should_end)
+        return self._export(BatchExportStrategy.EXPORT_ALL, deadline)

opentelemetry-sdk/tests/shared_internal/test_batch_processor.py

@@ -20,6 +20,7 @@
 import threading
 import time
 import unittest
+import unittest.mock
 import weakref
 from platform import system
 from typing import Any
@@ -194,32 +195,33 @@ def test_force_flush_returns_true_when_all_exported(
     def test_force_flush_returns_false_when_timeout_exceeded(
         self, batch_processor_class, telemetry
     ):
-        call_count = 0
-
-        def slow_export(batch):
-            nonlocal call_count
-            call_count += 1
-            # Sleep long enough that the deadline is exceeded after first batch.
-            time.sleep(0.2)
-
         exporter = Mock()
-        exporter.export.side_effect = slow_export
         batch_processor = batch_processor_class(
             exporter,
-            max_queue_size=200,
+            max_queue_size=15,
             max_export_batch_size=1,
-            # Long enough that the worker thread won't wake up during the test.
             schedule_delay_millis=30000,
             export_timeout_millis=500,
         )
-        for _ in range(50):
+        # Stop the worker thread first so it cannot export or interfere.
+        batch_processor._batch_processor._shutdown = True
+        batch_processor._batch_processor._worker_awaken.set()
+        batch_processor._batch_processor._worker_thread.join()
+        # Reset _shutdown so force_flush is not a no-op.
+        batch_processor._batch_processor._shutdown = False
+        # Emit items after worker is stopped.
+        for _ in range(3):
             batch_processor._batch_processor.emit(telemetry)
-        # 100ms timeout, each export takes 200ms, so deadline is hit after first batch.
-        result = batch_processor.force_flush(timeout_millis=100)
+        # Mock time.time(): first call computes deadline, second call is past it.
+        start = time.time()
+        with unittest.mock.patch(
+            "opentelemetry.sdk._shared_internal.time.time",
+            side_effect=[start, start + 1000],
+        ):
+            result = batch_processor.force_flush(timeout_millis=100)
         assert result is False
-        # Exporter was called at least once but not for all batches.
-        assert 1 <= call_count < 50
-        batch_processor.shutdown()
+        batch_processor._batch_processor._shutdown = True
+        batch_processor._batch_processor._exporter.shutdown()
 
     # pylint: disable=no-self-use
     def test_force_flush_returns_false_when_shutdown(