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 d5adff3c24d5f97319776d4ff0bd09ce6a06296f Author: Maciej Obuchowski <[email protected]> AuthorDate: Thu Jun 18 15:05:00 2026 +0200 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 <[email protected]> --- providers/openlineage/provider.yaml | 5 +++-- .../providers/openlineage/get_provider_info.py | 2 +- .../providers/openlineage/plugins/listener.py | 23 +++++++++++++--------- 3 files changed, 18 insertions(+), 12 deletions(-) diff --git a/providers/openlineage/provider.yaml b/providers/openlineage/provider.yaml index 51a4c02f027..4b48af3f318 100644 --- a/providers/openlineage/provider.yaml +++ b/providers/openlineage/provider.yaml @@ -192,8 +192,9 @@ config: (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. + 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. 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 d73e8a3106b..56b40466c2e 100644 --- a/providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py +++ b/providers/openlineage/src/airflow/providers/openlineage/get_provider_info.py @@ -107,7 +107,7 @@ def get_provider_info(): "version_added": "1.1.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 [...] + "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", diff --git a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py index 7be2ba00d95..82f90d6715d 100644 --- a/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py +++ b/providers/openlineage/src/airflow/providers/openlineage/plugins/listener.py @@ -833,11 +833,12 @@ class OpenLineageListener: 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. + 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. """ thread = threading.Thread( target=callable, @@ -847,10 +848,14 @@ class OpenLineageListener: 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. + # 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.",
