Honor execute_tasks_new_python_interpreter in the task supervisor

The config currently stops one fork short: the executor honors it when
launching the supervisor, but the Task SDK supervisor then bare-forks the
task runner anyway. On a long-running multithreaded supervisor (OTel
exporter, google-auth refresh threads, OpenLineage listener) that fork
can inherit OpenSSL's global lock mid-held by a sibling thread, and every
SSLContext construction in the child then blocks forever (issue #71707).

Reuse the existing fork+exec machinery (already used on macOS) so Linux
deployments can opt into a fresh interpreter per task with the same
[core] execute_tasks_new_python_interpreter knob, matching what the Edge
worker did in #65943.

Author: goingforstudying-ctrl

airflow-core/src/airflow/config_templates/config.yml

@@ -223,6 +223,11 @@ core:
         * ``False``: Execute via forking of the parent process
         * ``True``: Spawning a new python process, slower than fork, but means plugin changes picked
           up by tasks straight away
+
+        A fresh interpreter also avoids fork-safety hazards of a long-running,
+        multithreaded supervisor: a bare ``fork()`` can inherit C-library locks held
+        by a sibling thread at fork time (e.g. OpenSSL's, taken while building an
+        ``SSLContext``), which deadlocks later TLS setup in the task process.
       default: "False"
       example: ~
       version_added: 2.0.0

task-sdk/src/airflow/sdk/execution_time/supervisor.py

@@ -517,6 +517,18 @@ def _should_use_exec() -> bool:
     return sys.platform in _FORK_EXEC_PLATFORMS
 
 
+def _task_exec_configured() -> bool:
+    """
+    Whether ``[core] execute_tasks_new_python_interpreter`` asks for a fresh interpreter per task.
+
+    Opt-in fork-safety escape hatch: a bare ``os.fork()`` of a multithreaded
+    supervisor can inherit a C-library lock mid-held by a sibling thread
+    (OpenSSL's global lock, taken inside ``SSL_CTX_new``), which then blocks
+    every ``SSLContext`` construction in the child forever.
+    """
+    return conf.getboolean("core", "execute_tasks_new_python_interpreter", fallback=False)
+
+
 def _resolve_child_target(dotted: str) -> Callable[[], None]:
     """
     Resolve a ``module:qualname`` string to the callable the exec'd child runs.
@@ -1412,10 +1424,13 @@ def start(  # type: ignore[override]
         **kwargs,
     ) -> Self:
         """Fork and start a new subprocess to execute the given task."""
-        # Opt in to fork+exec on platforms that need it (currently macOS).
+        # Opt in to fork+exec on platforms that need it (currently macOS), or
+        # when a fresh interpreter per task is configured (fork of a
+        # multithreaded supervisor can inherit held C-library locks, e.g.
+        # OpenSSL's, and deadlock the child; see #71707).
         # Tests override `target` with a local stub to exercise the base
         # infrastructure; keep bare fork for those.
-        use_exec = target is _subprocess_main and _should_use_exec()
+        use_exec = target is _subprocess_main and (_should_use_exec() or _task_exec_configured())
         proc: Self = super().start(
             id=what.id,
             client=client,

task-sdk/tests/task_sdk/execution_time/test_supervisor.py

@@ -4479,6 +4479,66 @@ def test_api_client_clears_dag_bag_override_when_dag_is_none():
         in_process_api_server.cache_clear()
 
 
+class TestActivitySubprocessExecSelection:
+    """Which launch mode ``ActivitySubprocess.start()`` picks for the task runner."""
+
+    @pytest.fixture
+    def captured_kwargs(self, monkeypatch):
+        captured = {}
+
+        @classmethod
+        def fake_start(cls, **kwargs):
+            captured.update(kwargs)
+            return MagicMock(spec=supervisor.ActivitySubprocess)
+
+        monkeypatch.setattr(supervisor.WatchedSubprocess, "start", fake_start)
+        return captured
+
+    def _start(self, **kwargs):
+        args = {
+            "what": TaskInstance(
+                id="4d828a62-a417-4936-a7a6-2b3fabacecab",
+                task_id="b",
+                dag_id="c",
+                run_id="d",
+                try_number=1,
+                dag_version_id=uuid7(),
+                queue="default",
+            ),
+            "dag_rel_path": os.devnull,
+            "bundle_info": FAKE_BUNDLE,
+            "client": MagicMock(spec=sdk_client.Client),
+        }
+        args.update(kwargs)
+        return ActivitySubprocess.start(**args)
+
+    def test_bare_fork_by_default(self, monkeypatch, captured_kwargs):
+        monkeypatch.setattr(supervisor, "_should_use_exec", lambda: False)
+        with conf_vars({("core", "execute_tasks_new_python_interpreter"): "false"}):
+            self._start()
+        assert captured_kwargs["use_exec"] is False
+
+    def test_exec_when_config_requests_fresh_interpreter(self, monkeypatch, captured_kwargs):
+        """Linux + execute_tasks_new_python_interpreter=True still execs (issue #71707)."""
+        monkeypatch.setattr(supervisor, "_should_use_exec", lambda: False)
+        with conf_vars({("core", "execute_tasks_new_python_interpreter"): "true"}):
+            self._start()
+        assert captured_kwargs["use_exec"] is True
+
+    def test_exec_on_fork_unsafe_platform_regardless_of_config(self, monkeypatch, captured_kwargs):
+        monkeypatch.setattr(supervisor, "_should_use_exec", lambda: True)
+        with conf_vars({("core", "execute_tasks_new_python_interpreter"): "false"}):
+            self._start()
+        assert captured_kwargs["use_exec"] is True
+
+    def test_stub_target_keeps_bare_fork_even_when_config_set(self, monkeypatch, captured_kwargs):
+        """Tests override `target` with local stubs; those can't be rehydrated across exec."""
+        monkeypatch.setattr(supervisor, "_should_use_exec", lambda: False)
+        with conf_vars({("core", "execute_tasks_new_python_interpreter"): "true"}):
+            self._start(target=lambda: None)
+        assert captured_kwargs["use_exec"] is False
+
+
 class TestResolveChildTarget:
     """Test rehydrating the exec'd child's entry point from _AIRFLOW_CHILD_TARGET."""