You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by mo...@apache.org on 2026/07/22 12:39:19 UTC
(airflow) branch main updated: OpenLineage: add execute_in_thread to emit task events without forking (#68708)
This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 373ca8614d5 OpenLineage: add execute_in_thread to emit task events without forking (#68708)
373ca8614d5 is described below
commit 373ca8614d5e804db9efa1312650c0347d85fa04
Author: Maciej Obuchowski <ob...@gmail.com>
AuthorDate: Wed Jul 22 14:39:10 2026 +0200
OpenLineage: add execute_in_thread to emit task events without forking (#68708)
* OpenLineage: add execute_in_thread to emit task events without forking
By default the OpenLineage listener emits each task-level event from a
forked child process (os.fork() with no exec). That child inherits the
task runner's connection to the Airflow supervisor; if the child's event
emission blocks (e.g. a slow or unreachable lineage backend), the
inherited connection can be left in a state that prevents the task from
being marked complete, leaving it stuck in the `running` state.
Add an opt-in `[openlineage] execute_in_thread` option (default False).
When enabled, task-level emission runs in a time-bounded daemon thread
instead of forking: nothing is inherited, so a blocked emission can never
strand the task, and the task runner waits at most
`[openlineage] execution_timeout` for emission before proceeding.
Metadata extraction still runs in-process with full access to the task
runtime, so Operators whose extractors resolve Connections, Variables or
XComs keep working.
The default (fork) path is unchanged.
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
* Regenerate OpenLineage get_provider_info.py with execute_in_thread config
The update-providers-build-files prek hook regenerates get_provider_info.py
from provider.yaml. Adding execute_in_thread to provider.yaml requires the
generated file to be updated in sync.
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
* OpenLineage: correct execute_in_thread docs on supervisor-channel behavior
The thread path also reaches the Airflow supervisor during in-process
metadata extraction, through the shared SUPERVISOR_COMMS threading lock.
That lock prevents the byte interleaving that corrupted the protocol under
fork, but it does mean the task runner can briefly wait on the lock while an
abandoned emission thread finishes a round trip. Reword the config docs,
docstring, and inline comment to describe this accurately instead of
claiming emission can "never block" the task runner.
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
* OpenLineage: match generated get_provider_info.py option ordering
The update-providers-build-files generator preserves provider.yaml option
order (execution_timeout then execute_in_thread), so place the generated
entry accordingly to keep the file in sync with the generator output.
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
* catch exceptions on thread path
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
* add troubleshooting doc
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
---------
Signed-off-by: Maciej Obuchowski <ma...@datadoghq.com>
---
providers/openlineage/docs/troubleshooting.rst | 44 ++++++++++
providers/openlineage/provider.yaml | 20 +++++
.../src/airflow/providers/openlineage/conf.py | 6 ++
.../providers/openlineage/get_provider_info.py | 7 ++
.../providers/openlineage/plugins/listener.py | 53 +++++++++++-
.../unit/openlineage/plugins/test_listener.py | 98 ++++++++++++++++++++++
.../tests/unit/openlineage/test_conf.py | 17 ++++
7 files changed, 244 insertions(+), 1 deletion(-)
diff --git a/providers/openlineage/docs/troubleshooting.rst b/providers/openlineage/docs/troubleshooting.rst
index 9c358a6273e..56397c9644f 100644
--- a/providers/openlineage/docs/troubleshooting.rst
+++ b/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
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
diff --git a/providers/openlineage/provider.yaml b/providers/openlineage/provider.yaml
index 7635f4c814c..289b3e213db 100644
--- a/providers/openlineage/provider.yaml
+++ b/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.
diff --git a/providers/openlineage/src/airflow/providers/openlineage/conf.py b/providers/openlineage/src/airflow/providers/openlineage/conf.py
index d46dcc67cd1..23e940b59d2 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/conf.py
+++ b/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."""
diff --git a/providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py b/providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py
index 117981894b4..4805d056835 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py
+++ b/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 tha [...]
+ "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",
diff --git a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
index 6ce03e6f078..a37c34ef54f 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
@@ -19,6 +19,7 @@ from __future__ import annotations
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 @@ class OpenLineageListener:
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:
diff --git a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
index a4a31884a3c..9168052311f 100644
--- a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
+++ b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
@@ -2890,3 +2890,101 @@ class TestOpenLineageSelectiveEnableAirflow2:
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
diff --git a/providers/openlineage/tests/unit/openlineage/test_conf.py b/providers/openlineage/tests/unit/openlineage/test_conf.py
index 5ee497a6151..61e32d46d1b 100644
--- a/providers/openlineage/tests/unit/openlineage/test_conf.py
+++ b/providers/openlineage/tests/unit/openlineage/test_conf.py
@@ -30,6 +30,7 @@ from airflow.providers.openlineage.conf import (
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_SELECTIVE_ENABLE = "selective_enable"
_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",
(