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/06/18 12:32:53 UTC
(airflow) branch openlineage-execute-in-thread created (now 6c05e7503ba)
This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a change to branch openlineage-execute-in-thread
in repository https://gitbox.apache.org/repos/asf/airflow.git
at 6c05e7503ba OpenLineage: add execute_in_thread to emit task events without forking
This branch includes the following new commits:
new 6c05e7503ba OpenLineage: add execute_in_thread to emit task events without forking
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(airflow) 01/01: OpenLineage: add execute_in_thread to emit task events without forking
Posted by mo...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch openlineage-execute-in-thread
in repository https://gitbox.apache.org/repos/asf/airflow.git
commit 6c05e7503bafb0cf1401200f9d5100e2f1286099
Author: Maciej Obuchowski <ma...@datadoghq.com>
AuthorDate: Mon Jun 15 17:53:06 2026 +0200
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>
---
providers/openlineage/provider.yaml | 19 +++++
.../src/airflow/providers/openlineage/conf.py | 6 ++
.../providers/openlineage/plugins/listener.py | 37 +++++++++-
.../unit/openlineage/plugins/test_listener.py | 85 ++++++++++++++++++++++
.../tests/unit/openlineage/test_conf.py | 17 +++++
5 files changed, 163 insertions(+), 1 deletion(-)
diff --git a/providers/openlineage/provider.yaml b/providers/openlineage/provider.yaml
index c219b2390f2..51a4c02f027 100644
--- a/providers/openlineage/provider.yaml
+++ b/providers/openlineage/provider.yaml
@@ -182,6 +182,25 @@ 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, so nothing is inherited and the task runner can never be blocked past
+ ``[openlineage] execution_timeout`` by event emission.
+
+ 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 3a2ca201e53..5aa8c47ba05 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/plugins/listener.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py
index 2d4d74e828f..7be2ba00d95 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
@@ -819,10 +820,44 @@ 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 the supervisor connection (and every other inherited
+ resource) is left untouched -- a blocked emission can therefore never leave the
+ task stuck in the ``running`` state. Metadata extraction still runs in-process
+ with full access to the task runtime, so Operators whose extractors resolve
+ Connections, Variables or XComs keep working.
+ """
+ thread = threading.Thread(
+ target=callable,
+ 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 holding only its own backend connection (never the supervisor socket),
+ # so abandoning it cannot block the task runner or the worker -- it is reaped when
+ # the process exits. 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 edec29202c0..716db433427 100644
--- a/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
+++ b/providers/openlineage/tests/unit/openlineage/plugins/test_listener.py
@@ -2319,3 +2319,88 @@ 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()
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",
(