[release/v1.45.x-0.66bx] opentelemetry-sdk: fix thread leak bug in resource initialization logic

.changelog/5706.fixed

@@ -0,0 +1 @@
+`opentelemetry-sdk`: fix thread leak bug in resource initialization logic for long running resource detectors

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

@@ -743,12 +743,10 @@ def get_aggregated_resources(
     """
     detectors_merged_resource = initial_resource or Resource.create()
 
-    # The executor is not used as a context manager: exiting it would call
-    # `shutdown(wait=True)` and join still-running workers, so a detector that
-    # blocks longer than `timeout` would hold up the whole call regardless of
-    # `future.result(timeout=...)`. Shut down without waiting instead so the
-    # timeout actually bounds this call. A detector that outlives the timeout
-    # continues in its worker thread until detect() returns.
+    # A detector that exceeds `timeout` is skipped, but shutdown still joins
+    # its worker so no threads outlive this call. A detector that blocks
+    # indefinitely therefore blocks this call. `cancel_futures=True` drops
+    # detectors that have not started yet if a `raise_on_error` detector raises.
     executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
     try:
         futures = [executor.submit(detector.detect) for detector in detectors]
@@ -773,6 +771,6 @@ def get_aggregated_resources(
             finally:
                 detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
     finally:
-        executor.shutdown(wait=False, cancel_futures=True)
+        executor.shutdown(wait=True, cancel_futures=True)
 
     return detectors_merged_resource

opentelemetry-sdk/tests/resources/test_resources.py

@@ -7,6 +7,7 @@
 import platform
 import subprocess
 import sys
+import threading
 import time
 import unittest
 import uuid
@@ -472,14 +473,20 @@ def test_resource_detector_raise_error(self):
         resource_detector.raise_on_error = True
         self.assertRaises(Exception, get_aggregated_resources, [resource_detector])
 
-    def test_aggregated_resources_timeout_bounds_wait(self):
-        # A detector that takes longer than the timeout must not hold up the
-        # call beyond the timeout: the skip warning fires, but the wait is bounded.
+    def test_aggregated_resources_joins_detector_threads(self):
+        # A detector that takes longer than the timeout is skipped, but its
+        # worker thread must be joined before the call returns so no threads
+        # outlive resource creation.
+        worker_threads = []
+
+        def slow_detect():
+            worker_threads.append(threading.current_thread())
+            time.sleep(0.5)
+
         resource_detector = Mock(spec=ResourceDetector)
-        resource_detector.detect.side_effect = lambda: time.sleep(2)
+        resource_detector.detect.side_effect = slow_detect
         resource_detector.raise_on_error = False
 
-        start = time.time()
         with self.assertLogs(level=WARNING) as log_entry:
             self.assertEqual(
                 get_aggregated_resources(
@@ -489,10 +496,10 @@ def test_aggregated_resources_timeout_bounds_wait(self):
                 ),
                 _DEFAULT_RESOURCE,
             )
-        elapsed = time.time() - start
 
-        self.assertLess(elapsed, 1.0)
         self.assertIn("took longer than", log_entry.output[0])
+        self.assertEqual(len(worker_threads), 1)
+        self.assertFalse(worker_threads[0].is_alive())
 
     def test_resource_detector_is_not_process_dependent_by_default(self):
         self.assertFalse(DefaultResourceDetector().is_process_dependent())