Fix race condition in ConcurrentMultiSpanProcessor.shutdown()

Fixed error during shutdown: 'cannot schedule new futures after shutdow'.
Modified shutdown sequence to handle executor already being in shutdown state.
Added direct processor shutdown when executor is unavailable.
Added regression test for executor shutdown race condition.

Author: jomcgi

CHANGELOG.md

@@ -32,6 +32,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([#4406](https://github.com/open-telemetry/opentelemetry-python/pull/4406))
 - Fix env var error message for TraceLimits/SpanLimits
   ([#4458](https://github.com/open-telemetry/opentelemetry-python/pull/4458))
+- Fix race condition in ConcurrentMultiSpanProcessor shutdown
+  ([#4462](https://github.com/open-telemetry/opentelemetry-python/pull/4462))
 
 ## Version 1.30.0/0.51b0 (2025-02-03)
 

failing_command.py

@@ -0,0 +1,19 @@
+from opentelemetry import trace
+from opentelemetry.sdk.trace import (
+    ConcurrentMultiSpanProcessor,
+    TracerProvider,
+)
+from opentelemetry.sdk.trace.export import (
+    BatchSpanProcessor,
+    ConsoleSpanExporter,
+)
+
+span_proc = ConcurrentMultiSpanProcessor()
+span_proc.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
+
+trace.set_tracer_provider(TracerProvider())
+trace.get_tracer_provider().add_span_processor(span_proc)
+
+tracer = trace.get_tracer(__name__)
+with tracer.start_as_current_span("foo"):
+    print("Hello world!")

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

@@ -221,15 +221,18 @@ class ConcurrentMultiSpanProcessor(SpanProcessor):
     def __init__(self, num_threads: int = 2):
         # use a tuple to avoid race conditions when adding a new span and
         # iterating through it on "on_start" and "on_end".
-        self._span_processors = ()  # type: Tuple[SpanProcessor, ...]
+        self._span_processors: tuple[SpanProcessor] = ()
         self._lock = threading.Lock()
         self._executor = concurrent.futures.ThreadPoolExecutor(
             max_workers=num_threads
         )
+        self._is_shutdown: bool = False
 
     def add_span_processor(self, span_processor: SpanProcessor) -> None:
         """Adds a SpanProcessor to the list handled by this instance."""
         with self._lock:
+            if self._is_shutdown:
+                return
             self._span_processors += (span_processor,)
 
     def _submit_and_await(
@@ -239,9 +242,13 @@ def _submit_and_await(
         **kwargs: Any,
     ):
         futures = []
-        for sp in self._span_processors:
-            future = self._executor.submit(func(sp), *args, **kwargs)
-            futures.append(future)
+        with self._lock:
+            if self._is_shutdown:
+                return
+            for sp in self._span_processors:
+                future = self._executor.submit(func(sp), *args, **kwargs)
+                futures.append(future)
+
         for future in futures:
             future.result()
 
@@ -258,8 +265,16 @@ def on_end(self, span: "ReadableSpan") -> None:
         self._submit_and_await(lambda sp: sp.on_end, span)
 
     def shutdown(self) -> None:
-        """Shuts down all underlying span processors in parallel."""
-        self._submit_and_await(lambda sp: sp.shutdown)
+        """Shuts down all underlying span processors and the executor."""
+        with self._lock:
+            if self._is_shutdown:
+                return
+            self._is_shutdown = True
+
+            for sp in self._span_processors:
+                sp.shutdown()
+
+            self._executor.shutdown(wait=True)
 
     def force_flush(self, timeout_millis: int = 30000) -> bool:
         """Calls force_flush on all underlying span processors in parallel.
@@ -272,10 +287,14 @@ def force_flush(self, timeout_millis: int = 30000) -> bool:
             True if all span processors flushed their spans within the given
             timeout, False otherwise.
         """
-        futures = []
-        for sp in self._span_processors:  # type: SpanProcessor
-            future = self._executor.submit(sp.force_flush, timeout_millis)
-            futures.append(future)
+        with self._lock:
+            if self._is_shutdown:
+                return False
+
+            futures = []
+            for sp in self._span_processors:
+                future = self._executor.submit(sp.force_flush, timeout_millis)
+                futures.append(future)
 
         timeout_sec = timeout_millis / 1e3
         done_futures, not_done_futures = concurrent.futures.wait(

opentelemetry-sdk/tests/trace/test_span_processor.py

@@ -11,8 +11,8 @@
 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 # See the License for the specific language governing permissions and
 # limitations under the License.
-
 import abc
+import threading
 import time
 import typing
 import unittest
@@ -26,6 +26,10 @@
 from opentelemetry import trace as trace_api
 from opentelemetry.context import Context
 from opentelemetry.sdk import trace
+from opentelemetry.sdk.trace.export import (
+    BatchSpanProcessor,
+    ConsoleSpanExporter,
+)
 
 
 def span_event_start_fmt(span_processor_name, span_name):
@@ -314,3 +318,25 @@ def test_force_flush_late_by_span_processor(self):
         for mock_processor in mocks:
             self.assertEqual(1, mock_processor.force_flush.call_count)
         multi_processor.shutdown()
+
+    def test_executor_shutdown(self):
+        span_proc = trace.ConcurrentMultiSpanProcessor()
+        span_proc.add_span_processor(BatchSpanProcessor(ConsoleSpanExporter()))
+
+        provider = trace.TracerProvider()
+        provider.add_span_processor(span_proc)
+
+        tracer = provider.get_tracer(__name__)
+        with tracer.start_as_current_span("foo"):
+            pass
+
+        shutdown_thread = threading.Thread(target=span_proc._executor.shutdown) # pylint: disable=protected-access
+        shutdown_thread.start()
+
+        try:
+            span_proc.shutdown()
+            shutdown_thread.join()
+        except RuntimeError as unexpected_error:
+            self.fail(
+                f"Unexpected error raised during shutdown: {unexpected_error}"
+            )