From a731d3ff800563bcdec930f64f2d3db6e99c8a1f Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Fri, 16 Oct 2026 20:24:16 +0530
Subject: [PATCH 1/6] fix(sdk): run process-dependent resource detectors
 synchronously in fork handlers (#5789)

Refresh process-dependent resources synchronously inside _get_process_dependent_resource()
rather than delegating to get_aggregated_resources().

Delegating to get_aggregated_resources() spawns a ThreadPoolExecutor in the fork
handler's after_in_child hook. If concurrent.futures.thread was imported after
the provider, the executor's submit() call deadlocks trying to acquire
_global_shutdown_lock before concurrent.futures.thread's after_in_child hook has
a chance to reinitialize the lock.

Running the detectors synchronously eliminates thread pool worker creation and lock
acquisition in the newly forked child, preserving detector order, exception handling,
raise_on_error semantics, and attribute merging without deadlock hazards.

Fixes #5789.
---
 .changelog/5789.fixed                         |  1 +
 .../opentelemetry/sdk/resources/__init__.py   | 18 ++++++++---
 opentelemetry-sdk/tests/trace/test_trace.py   | 32 +++++++++++++++++++
 3 files changed, 47 insertions(+), 4 deletions(-)
 create mode 100644 .changelog/5789.fixed

diff --git a/.changelog/5789.fixed b/.changelog/5789.fixed
new file mode 100644
index 00000000000..d368c241a7f
--- /dev/null
+++ b/.changelog/5789.fixed
@@ -0,0 +1 @@
+`opentelemetry-sdk`: run process-dependent resource detectors synchronously in fork handlers to avoid post-fork deadlock when thread pool locks are held
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index 4c2a484b004..c4c2eaa743e 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -743,10 +743,20 @@ def _build_resource_detectors() -> list["ResourceDetector"]:
 
 
 def _get_process_dependent_resource() -> Resource:  # pyright: ignore[reportUnusedFunction]
-    return get_aggregated_resources(
-        [detector for detector in _build_resource_detectors() if detector.is_process_dependent()],
-        Resource.get_empty(),
-    )
+    detectors = [detector for detector in _build_resource_detectors() if detector.is_process_dependent()]
+    detectors_merged_resource = Resource.get_empty()
+    for detector in detectors:
+        detected_resource: Resource = _EMPTY_RESOURCE
+        try:
+            detected_resource = detector.detect()
+        # pylint: disable=broad-exception-caught
+        except Exception as ex:
+            if detector.raise_on_error:
+                raise ex
+            logger.warning("Exception %s in detector %s, ignoring", ex, detector)
+        finally:
+            detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
+    return detectors_merged_resource
 
 
 def get_aggregated_resources(
diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py
index f5eb7128909..d26a5ad0ba7 100644
--- a/opentelemetry-sdk/tests/trace/test_trace.py
+++ b/opentelemetry-sdk/tests/trace/test_trace.py
@@ -9,6 +9,7 @@
 import json
 import os
 import shutil
+import signal
 import subprocess
 import sys
 import unittest
@@ -210,6 +211,37 @@ def test_tracer_provider_updates_process_dependent_resource_after_fork(
         self.assertEqual(child_payload["new_tracer_pid"], child_payload["child_pid"])
         self.assertEqual(child_payload["span_pid"], child_payload["child_pid"])
 
+    @unittest.skipUnless(
+        hasattr(os, "fork") and hasattr(os, "register_at_fork"),
+        "requires os.fork and os.register_at_fork",
+    )
+    def test_tracer_provider_fork_with_preexec_fn_does_not_deadlock(self):
+        code = (
+            "import os, signal, subprocess, sys\n"
+            "from opentelemetry.sdk.resources import Resource\n"
+            "from opentelemetry.sdk.trace import TracerProvider\n"
+            "provider = TracerProvider(resource=Resource({'service.name': 'test'}))\n"
+            "import concurrent.futures.thread\n"
+            "proc = subprocess.Popen(\n"
+            "    ['true'], preexec_fn=lambda: signal.signal(signal.SIGPIPE, signal.SIG_DFL)\n"
+            ")\n"
+            "proc.wait()\n"
+            "sys.exit(proc.returncode)\n"
+        )
+        proc = subprocess.Popen(
+            [sys.executable, "-c", code],
+            start_new_session=True,
+            stdout=subprocess.PIPE,
+            stderr=subprocess.PIPE,
+        )
+        try:
+            stdout, stderr = proc.communicate(timeout=5)
+        except subprocess.TimeoutExpired:
+            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+            proc.communicate()
+            self.fail("TracerProvider fork handler deadlocked during fork with preexec_fn")
+        self.assertEqual(proc.returncode, 0, f"stdout: {stdout}\nstderr: {stderr}")
+
     def test_get_tracer_sdk(self):
         tracer_provider = trace.TracerProvider()
         tracer = tracer_provider.get_tracer(

From fe0214e74863a1c01ca0f3cfcb533b4fe9aebb4e Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Sat, 17 Oct 2026 01:19:46 +0530
Subject: [PATCH 2/6] fix(sdk): reuse resource aggregation in fork handlers

---
 .../opentelemetry/sdk/resources/__init__.py   | 37 ++++++++-----
 .../tests/resources/test_resources.py         | 55 +++++++++++++++++++
 2 files changed, 77 insertions(+), 15 deletions(-)

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index c4c2eaa743e..65034efc712 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -743,36 +743,43 @@ def _build_resource_detectors() -> list["ResourceDetector"]:
 
 
 def _get_process_dependent_resource() -> Resource:  # pyright: ignore[reportUnusedFunction]
-    detectors = [detector for detector in _build_resource_detectors() if detector.is_process_dependent()]
-    detectors_merged_resource = Resource.get_empty()
-    for detector in detectors:
-        detected_resource: Resource = _EMPTY_RESOURCE
-        try:
-            detected_resource = detector.detect()
-        # pylint: disable=broad-exception-caught
-        except Exception as ex:
-            if detector.raise_on_error:
-                raise ex
-            logger.warning("Exception %s in detector %s, ignoring", ex, detector)
-        finally:
-            detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
-    return detectors_merged_resource
+    return get_aggregated_resources(
+        [detector for detector in _build_resource_detectors() if detector.is_process_dependent()],
+        Resource.get_empty(),
+        use_threads=False,
+    )
 
 
 def get_aggregated_resources(
     detectors: Sequence["ResourceDetector"],
     initial_resource: Resource | None = None,
     timeout: int = 5,
+    use_threads: bool = True,
 ) -> "Resource":
     """Retrieves resources from detectors in the order that they were passed
 
     :param detectors: List of resources in order of priority
     :param initial_resource: Static resource. This has highest priority
-    :param timeout: Number of seconds to wait for each detector to return
+    :param timeout: Number of seconds to wait for each detector to return when running in a thread pool
+    :param use_threads: True to run detectors in a thread pool, False to run synchronously (timeouts do not apply in synchronous mode)
     :return:
     """
     detectors_merged_resource = initial_resource or Resource.create()
 
+    if not use_threads:
+        for detector in detectors:
+            detected_resource: Resource = _EMPTY_RESOURCE
+            try:
+                detected_resource = detector.detect()
+            # pylint: disable=broad-exception-caught
+            except Exception as ex:
+                if detector.raise_on_error:
+                    raise ex
+                logger.warning("Exception %s in detector %s, ignoring", ex, detector)
+            finally:
+                detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
+        return detectors_merged_resource
+
     # 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
diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py
index db67f9bbc5e..21674559872 100644
--- a/opentelemetry-sdk/tests/resources/test_resources.py
+++ b/opentelemetry-sdk/tests/resources/test_resources.py
@@ -557,6 +557,61 @@ def slow_detect():
         self.assertEqual(len(worker_threads), 1)
         self.assertFalse(worker_threads[0].is_alive())
 
+    def test_get_aggregated_resources_synchronous(self):
+        resource_detector1 = Mock(spec=ResourceDetector)
+        resource_detector1.detect.return_value = Resource({"key1": "val1", "shared": "old"})
+        resource_detector1.raise_on_error = False
+
+        resource_detector2 = Mock(spec=ResourceDetector)
+        resource_detector2.detect.return_value = Resource({"key2": "val2", "shared": "new"})
+        resource_detector2.raise_on_error = False
+
+        res = get_aggregated_resources(
+            [resource_detector1, resource_detector2],
+            initial_resource=Resource({"initial": "init"}),
+            use_threads=False,
+        )
+        self.assertEqual(
+            res,
+            Resource({"initial": "init", "key1": "val1", "key2": "val2", "shared": "new"}),
+        )
+
+    @patch("concurrent.futures.ThreadPoolExecutor")
+    def test_get_aggregated_resources_synchronous_does_not_instantiate_thread_pool(self, mock_executor):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.return_value = Resource({"k": "v"})
+        resource_detector.raise_on_error = False
+
+        res = get_aggregated_resources(
+            [resource_detector],
+            initial_resource=Resource.get_empty(),
+            use_threads=False,
+        )
+        self.assertEqual(res, Resource({"k": "v"}))
+        mock_executor.assert_not_called()
+
+    def test_get_aggregated_resources_synchronous_error_logged(self):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.side_effect = RuntimeError("detector failure")
+        resource_detector.raise_on_error = False
+
+        with self.assertLogs(level=WARNING) as log_entry:
+            res = get_aggregated_resources(
+                [resource_detector],
+                initial_resource=Resource({"k": "v"}),
+                use_threads=False,
+            )
+            self.assertEqual(res, Resource({"k": "v"}))
+        self.assertIn("Exception detector failure in detector", log_entry.output[0])
+
+    def test_get_aggregated_resources_synchronous_raise_error(self):
+        resource_detector = Mock(spec=ResourceDetector)
+        resource_detector.detect.side_effect = RuntimeError("fatal error")
+        resource_detector.raise_on_error = True
+
+        with self.assertRaises(RuntimeError):
+            get_aggregated_resources([resource_detector], use_threads=False)
+
     def test_resource_detector_is_not_process_dependent_by_default(self):
         self.assertFalse(DefaultResourceDetector().is_process_dependent())
 

From 88b84c8ca1f43c89b329f65efc23b356378ee04a Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Sun, 18 Oct 2026 09:57:53 +0530
Subject: [PATCH 3/6] refactor(sdk): generalize resource aggregation and
 preserve synchronous mode on default creation

Refactor get_aggregated_resources() to unify detector result resolution and error handling across both threaded and synchronous execution paths, eliminating duplicated loops. Additionally, pass `use_threads` to Resource.create() when initial_resource is omitted so synchronous aggregation does not spawn a thread pool.
---
 .../opentelemetry/sdk/resources/__init__.py   | 56 +++++++++----------
 .../tests/resources/test_resources.py         |  7 +++
 2 files changed, 35 insertions(+), 28 deletions(-)

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index 65034efc712..f007846d12e 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -54,6 +54,7 @@
 
 import abc
 import concurrent.futures
+import functools
 import logging
 import os
 import platform
@@ -182,6 +183,7 @@ def __init__(self, attributes: Attributes, schema_url: str | None = None):
     def create(
         attributes: Attributes | None = None,
         schema_url: str | None = None,
+        use_threads: bool = True,
     ) -> "Resource":
         """Creates a new `Resource` from attributes.
 
@@ -190,6 +192,7 @@ def create(
         Args:
             attributes: Optional zero or more key-value pairs.
             schema_url: Optional URL pointing to the schema
+            use_threads: Optional flag to run detectors in a thread pool or synchronously
 
         Returns:
             The newly-created Resource.
@@ -198,9 +201,9 @@ def create(
         if not attributes:
             attributes = {}
 
-        resource = get_aggregated_resources(_build_resource_detectors(), _DEFAULT_RESOURCE).merge(
-            Resource(attributes, schema_url)
-        )
+        resource = get_aggregated_resources(
+            _build_resource_detectors(), _DEFAULT_RESOURCE, use_threads=use_threads
+        ).merge(Resource(attributes, schema_url))
 
         if not resource.attributes.get(SERVICE_NAME, None):
             default_service_name = "unknown_service"
@@ -764,34 +767,30 @@ def get_aggregated_resources(
     :param use_threads: True to run detectors in a thread pool, False to run synchronously (timeouts do not apply in synchronous mode)
     :return:
     """
-    detectors_merged_resource = initial_resource or Resource.create()
-
-    if not use_threads:
-        for detector in detectors:
-            detected_resource: Resource = _EMPTY_RESOURCE
-            try:
-                detected_resource = detector.detect()
-            # pylint: disable=broad-exception-caught
-            except Exception as ex:
-                if detector.raise_on_error:
-                    raise ex
-                logger.warning("Exception %s in detector %s, ignoring", ex, detector)
-            finally:
-                detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
-        return detectors_merged_resource
+    detectors_merged_resource = (
+        initial_resource if initial_resource is not None else Resource.create(use_threads=use_threads)
+    )
 
-    # 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)
+    executor: concurrent.futures.ThreadPoolExecutor | None = None
     try:
-        futures = [executor.submit(detector.detect) for detector in detectors]
-        for detector_ind, future in enumerate(futures):
-            detector = detectors[detector_ind]
+        if use_threads:
+            # 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)
+            futures = [executor.submit(detector.detect) for detector in detectors]
+            tasks = [
+                (detector, functools.partial(future.result, timeout=timeout))
+                for detector, future in zip(detectors, futures)
+            ]
+        else:
+            tasks = [(detector, detector.detect) for detector in detectors]
+
+        for detector, get_result in tasks:
             detected_resource: Resource = _EMPTY_RESOURCE
             try:
-                detected_resource = future.result(timeout=timeout)
+                detected_resource = get_result()
             except concurrent.futures.TimeoutError as ex:
                 if detector.raise_on_error:
                     raise ex
@@ -808,6 +807,7 @@ def get_aggregated_resources(
             finally:
                 detectors_merged_resource = detectors_merged_resource.merge(detected_resource)
     finally:
-        executor.shutdown(wait=True, cancel_futures=True)
+        if executor is not None:
+            executor.shutdown(wait=True, cancel_futures=True)
 
     return detectors_merged_resource
diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py
index 21674559872..320dd999b5f 100644
--- a/opentelemetry-sdk/tests/resources/test_resources.py
+++ b/opentelemetry-sdk/tests/resources/test_resources.py
@@ -590,6 +590,13 @@ def test_get_aggregated_resources_synchronous_does_not_instantiate_thread_pool(s
         self.assertEqual(res, Resource({"k": "v"}))
         mock_executor.assert_not_called()
 
+        res_default = get_aggregated_resources(
+            [resource_detector],
+            use_threads=False,
+        )
+        self.assertEqual(res_default.attributes["k"], "v")
+        mock_executor.assert_not_called()
+
     def test_get_aggregated_resources_synchronous_error_logged(self):
         resource_detector = Mock(spec=ResourceDetector)
         resource_detector.detect.side_effect = RuntimeError("detector failure")

From 63d1be5963e14519334706ab268a2f795a4102d0 Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Tue, 20 Oct 2026 07:33:51 +0530
Subject: [PATCH 4/6] refactor(sdk): make thread pool option private and fix
 linter/changelog

- Make thread pool configuration private keyword-only (_use_thread_pool)
  in get_aggregated_resources() and Resource.create() per review feedback
- Use context manager for subprocess.Popen in fork deadlock regression test
  to resolve pylint R1732
- Rename changelog fragment to 5790.fixed to match pull request number
---
 .changelog/{5789.fixed => 5790.fixed}         |  0
 .../opentelemetry/sdk/resources/__init__.py   | 26 +++++++++----------
 .../tests/resources/test_resources.py         | 10 +++----
 opentelemetry-sdk/tests/trace/test_trace.py   | 18 ++++++-------
 4 files changed, 27 insertions(+), 27 deletions(-)
 rename .changelog/{5789.fixed => 5790.fixed} (100%)

diff --git a/.changelog/5789.fixed b/.changelog/5790.fixed
similarity index 100%
rename from .changelog/5789.fixed
rename to .changelog/5790.fixed
diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index f007846d12e..17ebbd68246 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -183,7 +183,8 @@ def __init__(self, attributes: Attributes, schema_url: str | None = None):
     def create(
         attributes: Attributes | None = None,
         schema_url: str | None = None,
-        use_threads: bool = True,
+        *,
+        _use_thread_pool: bool = True,
     ) -> "Resource":
         """Creates a new `Resource` from attributes.
 
@@ -192,7 +193,6 @@ def create(
         Args:
             attributes: Optional zero or more key-value pairs.
             schema_url: Optional URL pointing to the schema
-            use_threads: Optional flag to run detectors in a thread pool or synchronously
 
         Returns:
             The newly-created Resource.
@@ -202,7 +202,9 @@ def create(
             attributes = {}
 
         resource = get_aggregated_resources(
-            _build_resource_detectors(), _DEFAULT_RESOURCE, use_threads=use_threads
+            _build_resource_detectors(),
+            _DEFAULT_RESOURCE,
+            _use_thread_pool=_use_thread_pool,
         ).merge(Resource(attributes, schema_url))
 
         if not resource.attributes.get(SERVICE_NAME, None):
@@ -749,7 +751,7 @@ def _get_process_dependent_resource() -> Resource:  # pyright: ignore[reportUnus
     return get_aggregated_resources(
         [detector for detector in _build_resource_detectors() if detector.is_process_dependent()],
         Resource.get_empty(),
-        use_threads=False,
+        _use_thread_pool=False,
     )
 
 
@@ -757,27 +759,25 @@ def get_aggregated_resources(
     detectors: Sequence["ResourceDetector"],
     initial_resource: Resource | None = None,
     timeout: int = 5,
-    use_threads: bool = True,
+    *,
+    _use_thread_pool: bool = True,
 ) -> "Resource":
     """Retrieves resources from detectors in the order that they were passed
 
     :param detectors: List of resources in order of priority
     :param initial_resource: Static resource. This has highest priority
-    :param timeout: Number of seconds to wait for each detector to return when running in a thread pool
-    :param use_threads: True to run detectors in a thread pool, False to run synchronously (timeouts do not apply in synchronous mode)
+    :param timeout: Number of seconds to wait for each detector to return
     :return:
     """
     detectors_merged_resource = (
-        initial_resource if initial_resource is not None else Resource.create(use_threads=use_threads)
+        initial_resource
+        if initial_resource is not None
+        else Resource.create(_use_thread_pool=_use_thread_pool)
     )
 
     executor: concurrent.futures.ThreadPoolExecutor | None = None
     try:
-        if use_threads:
-            # 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.
+        if _use_thread_pool:
             executor = concurrent.futures.ThreadPoolExecutor(max_workers=4)
             futures = [executor.submit(detector.detect) for detector in detectors]
             tasks = [
diff --git a/opentelemetry-sdk/tests/resources/test_resources.py b/opentelemetry-sdk/tests/resources/test_resources.py
index 320dd999b5f..47b3ed5c3eb 100644
--- a/opentelemetry-sdk/tests/resources/test_resources.py
+++ b/opentelemetry-sdk/tests/resources/test_resources.py
@@ -569,7 +569,7 @@ def test_get_aggregated_resources_synchronous(self):
         res = get_aggregated_resources(
             [resource_detector1, resource_detector2],
             initial_resource=Resource({"initial": "init"}),
-            use_threads=False,
+            _use_thread_pool=False,
         )
         self.assertEqual(
             res,
@@ -585,14 +585,14 @@ def test_get_aggregated_resources_synchronous_does_not_instantiate_thread_pool(s
         res = get_aggregated_resources(
             [resource_detector],
             initial_resource=Resource.get_empty(),
-            use_threads=False,
+            _use_thread_pool=False,
         )
         self.assertEqual(res, Resource({"k": "v"}))
         mock_executor.assert_not_called()
 
         res_default = get_aggregated_resources(
             [resource_detector],
-            use_threads=False,
+            _use_thread_pool=False,
         )
         self.assertEqual(res_default.attributes["k"], "v")
         mock_executor.assert_not_called()
@@ -606,7 +606,7 @@ def test_get_aggregated_resources_synchronous_error_logged(self):
             res = get_aggregated_resources(
                 [resource_detector],
                 initial_resource=Resource({"k": "v"}),
-                use_threads=False,
+                _use_thread_pool=False,
             )
             self.assertEqual(res, Resource({"k": "v"}))
         self.assertIn("Exception detector failure in detector", log_entry.output[0])
@@ -617,7 +617,7 @@ def test_get_aggregated_resources_synchronous_raise_error(self):
         resource_detector.raise_on_error = True
 
         with self.assertRaises(RuntimeError):
-            get_aggregated_resources([resource_detector], use_threads=False)
+            get_aggregated_resources([resource_detector], _use_thread_pool=False)
 
     def test_resource_detector_is_not_process_dependent_by_default(self):
         self.assertFalse(DefaultResourceDetector().is_process_dependent())
diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py
index d26a5ad0ba7..79159b4189d 100644
--- a/opentelemetry-sdk/tests/trace/test_trace.py
+++ b/opentelemetry-sdk/tests/trace/test_trace.py
@@ -228,19 +228,19 @@ def test_tracer_provider_fork_with_preexec_fn_does_not_deadlock(self):
             "proc.wait()\n"
             "sys.exit(proc.returncode)\n"
         )
-        proc = subprocess.Popen(
+        with subprocess.Popen(
             [sys.executable, "-c", code],
             start_new_session=True,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
-        )
-        try:
-            stdout, stderr = proc.communicate(timeout=5)
-        except subprocess.TimeoutExpired:
-            os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-            proc.communicate()
-            self.fail("TracerProvider fork handler deadlocked during fork with preexec_fn")
-        self.assertEqual(proc.returncode, 0, f"stdout: {stdout}\nstderr: {stderr}")
+        ) as proc:
+            try:
+                stdout, stderr = proc.communicate(timeout=5)
+            except subprocess.TimeoutExpired:
+                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
+                proc.communicate()
+                self.fail("TracerProvider fork handler deadlocked during fork with preexec_fn")
+            self.assertEqual(proc.returncode, 0, f"stdout: {stdout}\nstderr: {stderr}")
 
     def test_get_tracer_sdk(self):
         tracer_provider = trace.TracerProvider()

From bcb16261c5362ae56c9d9220725453bca9965d40 Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Tue, 20 Oct 2026 21:23:45 +0530
Subject: [PATCH 5/6] test(trace): move fork preexec test logic to scripts
 helper and reformat

---
 .../opentelemetry/sdk/resources/__init__.py   |  4 +-
 .../scripts/tracer_provider_fork_preexec.py   | 23 +++++++++++
 opentelemetry-sdk/tests/trace/test_trace.py   | 39 +++++++------------
 3 files changed, 37 insertions(+), 29 deletions(-)
 create mode 100644 opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py

diff --git a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
index 17ebbd68246..1e7397f5c2f 100644
--- a/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
+++ b/opentelemetry-sdk/src/opentelemetry/sdk/resources/__init__.py
@@ -770,9 +770,7 @@ def get_aggregated_resources(
     :return:
     """
     detectors_merged_resource = (
-        initial_resource
-        if initial_resource is not None
-        else Resource.create(_use_thread_pool=_use_thread_pool)
+        initial_resource if initial_resource is not None else Resource.create(_use_thread_pool=_use_thread_pool)
     )
 
     executor: concurrent.futures.ThreadPoolExecutor | None = None
diff --git a/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py b/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py
new file mode 100644
index 00000000000..2988d09fd40
--- /dev/null
+++ b/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py
@@ -0,0 +1,23 @@
+# Copyright The OpenTelemetry Authors
+# SPDX-License-Identifier: Apache-2.0
+
+import signal
+import subprocess
+import sys
+
+from opentelemetry.sdk.resources import Resource
+from opentelemetry.sdk.trace import TracerProvider
+
+
+def main() -> None:
+    _provider = TracerProvider(resource=Resource({"service.name": "test"}))
+    proc = subprocess.Popen(
+        ["true"],
+        preexec_fn=lambda: signal.signal(signal.SIGPIPE, signal.SIG_DFL),  # noqa: PLW1509
+    )
+    proc.wait()
+    sys.exit(proc.returncode)
+
+
+if __name__ == "__main__":
+    main()
diff --git a/opentelemetry-sdk/tests/trace/test_trace.py b/opentelemetry-sdk/tests/trace/test_trace.py
index 79159b4189d..0703d6a3647 100644
--- a/opentelemetry-sdk/tests/trace/test_trace.py
+++ b/opentelemetry-sdk/tests/trace/test_trace.py
@@ -9,7 +9,6 @@
 import json
 import os
 import shutil
-import signal
 import subprocess
 import sys
 import unittest
@@ -216,31 +215,19 @@ def test_tracer_provider_updates_process_dependent_resource_after_fork(
         "requires os.fork and os.register_at_fork",
     )
     def test_tracer_provider_fork_with_preexec_fn_does_not_deadlock(self):
-        code = (
-            "import os, signal, subprocess, sys\n"
-            "from opentelemetry.sdk.resources import Resource\n"
-            "from opentelemetry.sdk.trace import TracerProvider\n"
-            "provider = TracerProvider(resource=Resource({'service.name': 'test'}))\n"
-            "import concurrent.futures.thread\n"
-            "proc = subprocess.Popen(\n"
-            "    ['true'], preexec_fn=lambda: signal.signal(signal.SIGPIPE, signal.SIG_DFL)\n"
-            ")\n"
-            "proc.wait()\n"
-            "sys.exit(proc.returncode)\n"
-        )
-        with subprocess.Popen(
-            [sys.executable, "-c", code],
-            start_new_session=True,
-            stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE,
-        ) as proc:
-            try:
-                stdout, stderr = proc.communicate(timeout=5)
-            except subprocess.TimeoutExpired:
-                os.killpg(os.getpgid(proc.pid), signal.SIGKILL)
-                proc.communicate()
-                self.fail("TracerProvider fork handler deadlocked during fork with preexec_fn")
-            self.assertEqual(proc.returncode, 0, f"stdout: {stdout}\nstderr: {stderr}")
+        script_path = Path(__file__).parent / "scripts" / "tracer_provider_fork_preexec.py"
+        result = subprocess.run(
+            [sys.executable, str(script_path)],
+            capture_output=True,
+            check=False,
+            text=True,
+            timeout=5,
+        )
+        self.assertEqual(
+            result.returncode,
+            0,
+            f"stdout: {result.stdout}\nstderr: {result.stderr}",
+        )
 
     def test_get_tracer_sdk(self):
         tracer_provider = trace.TracerProvider()

From 0f33eb8e62e03479819f567b53a7d322b5ed42e4 Mon Sep 17 00:00:00 2001
From: Chirag Jangir <184008125+chiragjangiir@users.noreply.github.com>
Date: Wed, 21 Oct 2026 11:07:16 +0530
Subject: [PATCH 6/6] fix(sdk): address pylint W1509 and R1732 in fork test
 helper

---
 .../tests/trace/scripts/tracer_provider_fork_preexec.py  | 9 +++++----
 1 file changed, 5 insertions(+), 4 deletions(-)

diff --git a/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py b/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py
index 2988d09fd40..cd575dbaa53 100644
--- a/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py
+++ b/opentelemetry-sdk/tests/trace/scripts/tracer_provider_fork_preexec.py
@@ -11,12 +11,13 @@
 
 def main() -> None:
     _provider = TracerProvider(resource=Resource({"service.name": "test"}))
-    proc = subprocess.Popen(
+    # pylint: disable=subprocess-popen-preexec-fn
+    with subprocess.Popen(
         ["true"],
         preexec_fn=lambda: signal.signal(signal.SIGPIPE, signal.SIG_DFL),  # noqa: PLW1509
-    )
-    proc.wait()
-    sys.exit(proc.returncode)
+    ) as proc:
+        proc.wait()
+        sys.exit(proc.returncode)
 
 
 if __name__ == "__main__":