OpenLineage: add execute_in_thread to emit task events without forking

[mobuchowski]

Why

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 there are issues with that connection - it can hang.

There were a bunch of issues with that fixed in 3.2.x release - like #66574 #65714 #66573 #67115 #66572 - but we still support earlier ones like 3.1.x line.

What

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.

Verified on AWS MWAA and custom GKE Airflow environment.

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: [Claude Code] following the guidelines

[kacpermuda]

The PR adds a well-motivated opt-in thread-based emission path that cleanly solves the supervisor-connection hang for most cases, and the default fork path is unchanged. Two nits (see inline) document genuine semantic differences between fork and thread isolation that slightly qualify the "can never block" guarantee in the docstring — but neither causes a runtime failure under normal conditions.

[kacpermuda]

Can we add some small paragraph about this in providers/openlineage/docs/troubleshooting.rst ? Can be helpful if we ever hit that again, to know what config to use. Also, let's clearly mention it there that it was observed on AF3.1, and may not be the case on other airflow versions since 3.2 has applied some fixes already

[kacpermuda]

Left few nit comments, looks solid otherwise - clean opt-in allowing a different execution mode, that can prevent some rare bug.

[abhipalsingh]

Adding a data point from the API server side, since this fork behavior isn't only a scheduler concern.

On Airflow 3.2.1 with apache-airflow-providers-openlineage==2.14.0 (pre-#65677), manual task-instance state changes via the REST
API (mark success/failed/skipped, clear) fire on_task_instance_* on the api-server — a multithreaded async process
(FastAPI/uvicorn under gunicorn). The listener's _fork_execute calls os.fork() from that multithreaded worker; a fraction of
children deadlock immediately on an inherited lock (py-spy showed them parked in futex_wait_queue, never reaching the post-fork setproctitle) and are never reaped → ~350–400 MB each → unbounded accumulation → api-server OOM.

#65677 helps the manual-state-change path (routes it through the ProcessPoolExecutor instead of a raw fork), but (a) that still forks a pool from the multithreaded async server, and (b) the natural-lifecycle handlers still use use_fork=True. So thread-based emission (this issue) is the cleaner fit for async contexts like the api-server, where os.fork() is fundamentally
unsafe.

We worked around it by disabling OpenLineage on the api-server (no transport configured there anyway), but big +1 for
execute_in_thread as the general fix.

[mobuchowski]

@abhipalsingh we should not run any extraction on API server, so I think fix should be separate - to not fork anyway; but do it not behind the configuration flag.

I think it should be closer to what we do in the scheduler with ProcessPoolExecutor maybe.