OpenLineage: add execute_in_thread to emit task events without forking

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (

providers/openlineage/docs/troubleshooting.rst

@@ -43,6 +43,50 @@ as well as the `task_success_overtime <https://airflow.apache.org/docs/apache-ai
 configuration in Airflow config.
 
 
+Tasks remain in the running state after finishing
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Airflow 3.1, OpenLineage task-level event emission can leave a task in the ``running`` state after its task code
+has finished. This can happen when the default fork-based emission path inherits the task runner's connection to
+the Airflow supervisor. Airflow 3.2 includes fixes for related supervisor communication issues, so this symptom may
+not occur on later Airflow versions.
+
+**Possible Solution**
+
+Upgrade to Airflow 3.2 or later where possible. On Airflow 3.1, enable
+:ref:`Execute In Thread <config:openlineage__execute_in_thread>` to run task-level event emission in a time-bounded
+thread instead of a forked child process:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+
+Fork DeprecationWarning on Python 3.12 and later
+^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
+
+On Python 3.12 and later, task-level OpenLineage event emission using the default fork-based path can produce the
+following warning:
+
+.. code-block:: text
+
+ DeprecationWarning: This process is multi-threaded, use of fork() may lead to deadlocks in the child.
+
+This warning indicates that ``fork()`` was called from a multi-threaded process.
+
+**Possible Solution**
+
+Enable :ref:`Execute In Thread <config:openlineage__execute_in_thread>` to avoid forking during task-level event
+emission:
+
+.. code-block:: ini
+
+ [openlineage]
+ execute_in_thread = True
+
+This setting does not change scheduler-side event emission performed by the process pool.
+
 Scheduler CPU or memory growing steadily while OpenLineage is enabled
 ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
 

providers/openlineage/provider.yaml

@@ -197,6 +197,26 @@ config:
  example: ~
  type: integer
  version_added: 1.9.0
+ execute_in_thread:
+ description: |
+ If true, OpenLineage task-level event emission on the worker runs in a time-bounded
+ background thread instead of a forked child process (the default).
+
+ The default fork model duplicates the task runner process - including its connection to
+ the Airflow supervisor - into a short-lived child. If the child's event emission blocks
+ (for example on a slow or unreachable lineage backend), the inherited supervisor
+ connection can be left in a state that prevents the task from being marked complete,
+ leaving it stuck in the ``running`` state. Running emission in a thread avoids forking
+ entirely: no supervisor connection is duplicated, so emission cannot strand the task in
+ the ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``
+ for emission and then proceeds, abandoning a thread that is still running.
+
+ Metadata extraction still runs in-process with full access to the task runtime, so
+ Operators whose extractors resolve Connections, Variables or XComs continue to work.
+ default: "False"
+ example: ~
+ type: boolean
+ version_added: 2.19.0
  extractors:
  description: |
  Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.

providers/openlineage/src/airflow/providers/openlineage/conf.py

@@ -169,6 +169,12 @@ def execution_timeout() -> int:
  return conf.getint(_CONFIG_SECTION, "execution_timeout", fallback="10")
 
 
+@cache
+def execute_in_thread() -> bool:
+ """[openlineage] execute_in_thread."""
+ return conf.getboolean(_CONFIG_SECTION, "execute_in_thread", fallback="False")
+
+
 @cache
 def include_full_task_info() -> bool:
  """[openlineage] include_full_task_info."""

providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py

@@ -120,6 +120,13 @@ def get_provider_info():
  "type": "integer",
  "version_added": "1.9.0",
  },
+ "execute_in_thread": {
+ "description": "If true, OpenLineage task-level event emission on the worker runs in a time-bounded\nbackground thread instead of a forked child process (the default).\n\nThe default fork model duplicates the task runner process - including its connection to\nthe Airflow supervisor - into a short-lived child. If the child's event emission blocks\n(for example on a slow or unreachable lineage backend), the inherited supervisor\nconnection can be left in a state that prevents the task from being marked complete,\nleaving it stuck in the ``running`` state. Running emission in a thread avoids forking\nentirely: no supervisor connection is duplicated, so emission cannot strand the task in\nthe ``running`` state. The task runner waits at most ``[openlineage] execution_timeout``\nfor emission and then proceeds, abandoning a thread that is still running.\n\nMetadata extraction still runs in-process with full access to the task runtime, so\nOperators whose extractors resolve Connections, Variables or XComs continue to work.\n",
+ "default": "False",
+ "example": None,
+ "type": "boolean",
+ "version_added": "2.19.0",
+ },
  "extractors": {
  "description": "Register custom OpenLineage Extractors by passing a string of semicolon separated full import paths.\n",
  "type": "string",

providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py

@@ -19,6 +19,7 @@
 import logging
 import os
 import sys
+import threading
 from concurrent.futures import ProcessPoolExecutor
 from concurrent.futures.process import BrokenProcessPool
 from datetime import datetime
@@ -974,10 +975,60 @@ def _on_task_instance_manual_state_change(
 
  def _execute(self, callable, callable_name: str, use_fork: bool = False):
  if use_fork:
- self._fork_execute(callable, callable_name)
+ if conf.execute_in_thread():
+ self._thread_execute(callable, callable_name)
+ else:
+ self._fork_execute(callable, callable_name)
  else:
  callable()
 
+ def _thread_execute(self, callable, callable_name: str):
+ """
+ Run OpenLineage event emission in a time-bounded daemon thread.
+
+ Opt-in alternative to :meth:`_fork_execute`, enabled via
+ ``[openlineage] execute_in_thread``. Unlike forking, this never duplicates the
+ task runner process, so no supervisor connection is inherited and left in a broken
+ state -- emission therefore cannot strand the task in the ``running`` state. The
+ task runner waits at most ``[openlineage] execution_timeout`` for emission and then
+ proceeds. Metadata extraction still runs in-process with full access to the task
+ runtime, so Operators whose extractors resolve Connections, Variables or XComs keep
+ working.
+ """
+
+ def _run():
+ try:
+ callable()
+ except Exception:
+ self.log.warning(
+ "OpenLineage %s thread failed. This has no impact on actual task execution status.",
+ callable_name,
+ exc_info=True,
+ )
+
+ thread = threading.Thread(
+ target=_run,
+ name=f"openlineage-{callable_name}",
+ daemon=True,
+ )
+ thread.start()
+ thread.join(timeout=conf.execution_timeout())
+ if thread.is_alive():
+ # Emission is still running. We deliberately do not keep waiting: the thread is a
+ # daemon, reaped when the process exits. Unlike the fork path -- where parent and
+ # child shared a socket fd with no cross-process locking and could interleave bytes
+ # on the supervisor channel -- this thread reaches the supervisor only through the
+ # shared SUPERVISOR_COMMS threading lock, so it cannot corrupt the protocol. The main
+ # thread may briefly wait on that lock if the abandoned thread is mid-request, but the
+ # wait is bounded by a single round trip. This mirrors the fork path terminating an
+ # over-running child.
+ self.log.warning(
+ "OpenLineage %s thread did not finish within execution_timeout=%ss and will be "
+ "abandoned. This has no impact on actual task execution status.",
+ callable_name,
+ conf.execution_timeout(),
+ )
+
  def _terminate_with_wait(self, process: psutil.Process):
  process.terminate()
  try:

providers/openlineage/tests/unit/openlineage/plugins/test_listener.py

@@ -2890,3 +2890,101 @@ def test_listener_with_dag_disabled_task_enabled(
 
  assert expected_call_count == listener._executor.submit.call_count
  assert expected_task_call_count == listener.extractor_manager.extract_metadata.call_count
+
+
+class TestExecuteRouting:
+ """Tests for `_execute` fork/thread routing and the `_thread_execute` bound (#65714 follow-up)."""
+
+ @conf_vars({("openlineage", "execute_in_thread"): "False"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_fork_by_default(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_fork.assert_called_once_with(callable_, "on_running")
+ mock_thread.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_uses_thread_when_enabled(self, mock_fork, mock_thread):
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_running", use_fork=True)
+
+ mock_thread.assert_called_once_with(callable_, "on_running")
+ mock_fork.assert_not_called()
+ callable_.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True"})
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._thread_execute")
+ @patch("airflow.providers.openlineage.plugins.listener.OpenLineageListener._fork_execute")
+ def test_execute_runs_inline_without_fork_flag(self, mock_fork, mock_thread):
+ # use_fork=False always runs the callable directly, regardless of execute_in_thread.
+ listener = OpenLineageListener()
+ callable_ = MagicMock()
+
+ listener._execute(callable_, "on_state_change", use_fork=False)
+
+ callable_.assert_called_once_with()
+ mock_fork.assert_not_called()
+ mock_thread.assert_not_called()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_runs_callable_to_completion(self):
+ listener = OpenLineageListener()
+ result = {"ran": False}
+
+ def _emit():
+ result["ran"] = True
+
+ listener._thread_execute(_emit, "on_running")
+
+ assert result["ran"] is True
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "1"})
+ def test_thread_execute_is_bounded_and_abandons_overrunning_emission(self):
+ import threading
+ import time
+
+ listener = OpenLineageListener()
+ started = threading.Event()
+ release = threading.Event()
+ finished = {"v": False}
+
+ def _slow_emit():
+ started.set()
+ # Block far longer than execution_timeout; released by the test for cleanup.
+ release.wait(timeout=30)
+ finished["v"] = True
+
+ try:
+ t0 = time.monotonic()
+ listener._thread_execute(_slow_emit, "on_running")
+ elapsed = time.monotonic() - t0
+
+ # The emission started, but `_thread_execute` returned without waiting for it:
+ # bounded by execution_timeout (1s), never blocking for the full 30s emission.
+ assert started.is_set()
+ assert finished["v"] is False
+ assert elapsed < 10
+ finally:
+ release.set()
+
+ @conf_vars({("openlineage", "execute_in_thread"): "True", ("openlineage", "execution_timeout"): "5"})
+ def test_thread_execute_swallows_callable_exception(self):
+ listener = OpenLineageListener()
+ listener.log = MagicMock()
+
+ def _failing_emit():
+ raise RuntimeError("boom")
+
+ listener._thread_execute(_failing_emit, "on_running")
+
+ listener.log.warning.assert_called_once()
+ assert "on_running" in listener.log.warning.call_args.args

providers/openlineage/tests/unit/openlineage/test_conf.py

@@ -30,6 +30,7 @@
  dag_state_change_process_pool_size,
  debug_mode,
  disabled_operators,
+ execute_in_thread,
  execution_timeout,
  include_full_task_info,
  is_disabled,
@@ -63,6 +64,7 @@
 _CONFIG_OPTION_DAG_STATE_CHANGE_PROCESS_POOL_SIZE = "dag_state_change_process_pool_size"
 _CONFIG_OPTION_INCLUDE_FULL_TASK_INFO = "include_full_task_info"
 _CONFIG_OPTION_DEBUG_MODE = "debug_mode"
+_CONFIG_OPTION_EXECUTE_IN_THREAD = "execute_in_thread"
 _CONFIG_OPTION_SPARK_INJECT_PARENT_JOB_INFO = "spark_inject_parent_job_info"
 _CONFIG_OPTION_SPARK_INJECT_TRANSPORT_INFO = "spark_inject_transport_info"
 
@@ -629,6 +631,21 @@ def test_debug_mode(var_string, expected):
  assert result is expected
 
 
+@pytest.mark.parametrize(
+ ("var_string", "expected"),
+ _BOOL_PARAMS,
+)
+def test_execute_in_thread(var_string, expected):
+ with conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): var_string}):
+ result = execute_in_thread()
+ assert result is expected
+
+
+@conf_vars({(_CONFIG_SECTION, _CONFIG_OPTION_EXECUTE_IN_THREAD): None})
+def test_execute_in_thread_do_not_fail_if_conf_option_missing():
+ assert execute_in_thread() is False
+
+
 @pytest.mark.parametrize(
  "var_string",
  (