OpenLineage: reuse per-process adapter in dag-state-change pool workers

[gang-zh]

Summary

The OpenLineage listener submits bound adapter methods to its DAG-run event ProcessPoolExecutor (self.adapter.dag_started / dag_success / dag_failed, and self.adapter.fail_task / complete_task via _emit_manual_state_change_event). Pickling a bound method serializes the adapter itself, so the long-lived pool worker unpickles a fresh OpenLineageAdapter for every event and, on emit, builds a new OpenLineageClient — with a new transport set — per DAG-run state change (get_or_create_openlineage_client caches on self._client, an instance attribute). close() is never called on those clients.

For transports that start background worker threads this leaks one thread per event inside the scheduler. The datadog transport (openlineage-python >= 1.37.0) unconditionally starts an AsyncHttpTransport worker thread in its constructor; each idle thread busy-polls at ~100 Hz and costs ~0.45% CPU (measured on openlineage-python 1.47.1 — see repro below; 400 accumulated clients = 401 threads = ~167% CPU doing nothing). Observed in production (Airflow 2.11, Astronomer, KubernetesExecutor, ~50 hourly DAGs, composite transport with a datadog leg): scheduler CPU climbs steadily from the moment the datadog leg is enabled until pinned at its limit within ~6 hours, heartbeat dips, and only a scheduler restart recovers it. Even the thread-less http transport pays a per-event session/connection-pool rebuild.

Fix

Route pool submissions through a module-level _run_adapter_method(method_name, ...) that resolves the adapter method by name on a per-process adapter singleton (_get_process_adapter), so each pool worker keeps exactly one client — and one transport set — for its whole lifetime. _emit_manual_state_change_event now takes the adapter method name and resolves it the same way.

Changes

• listener.py: add _get_process_adapter() / _run_adapter_method(); DAG-run hooks submit method names instead of bound methods; _emit_manual_state_change_event resolves the named method on the per-process adapter
• test_listener.py: regression test asserting two submissions executed in the same pool worker observe the same adapter instance; direct_submit_call test stand-in resolves method names on the listener's adapter so existing mocked-adapter assertions keep working
• docs/troubleshooting.rst: known-limitations entry for the steadily-growing scheduler CPU/memory symptom on affected versions

Design notes

• The fix is scoped to the pool boundary in the listener rather than making the adapter's client cache global: tests (and potentially users) construct adapters with injected per-instance clients, and the fork-based task-event path on workers intentionally uses the in-process adapter — both keep their existing semantics.
• No locking is needed in _get_process_adapter: pool workers execute submitted tasks serially, and the scheduler parent only submits (it never runs _run_adapter_method itself).

Test Plan

• New regression test test_process_adapter_reused_across_pool_submissions passes
• Full test_listener.py suite on main: 38 passed, 0 failed (38 Airflow-2 variants skip locally and run in the CI compatibility matrix)
• ruff check and ruff format --check clean on changed files

Thread-leak repro (openlineage-python 1.47.1) — one DatadogTransport instantiation per simulated event

import threading
import time

from openlineage.client.transport.datadog import DatadogConfig, DatadogTransport


def cpu_pct(interval: float = 3.0) -> float:
    t0, w0 = time.process_time(), time.perf_counter()
    time.sleep(interval)
    return 100 * (time.process_time() - t0) / (time.perf_counter() - w0)


transports = []
print(f"{'clients':>8} {'threads':>8} {'idle CPU%':>10}")
print(f"{0:>8} {threading.active_count():>8} {cpu_pct():>10.1f}")
for target in (50, 150, 400):
    while len(transports) < target:
        # one DAG-run state change == one fresh client == one DatadogTransport
        transports.append(DatadogTransport(DatadogConfig(apiKey="fake-key")))
    time.sleep(1.0)
    print(f"{len(transports):>8} {threading.active_count():>8} {cpu_pct():>10.1f}")

clients created (= events)	threads	idle CPU%
0	1	0.0
50	51	29.5
150	151	72.0
400	401	167.6

A complementary report about the datadog transport itself (unconditional thread start, never closed, busy-poll idle loop) is being filed with OpenLineage separately; this provider-side fix removes the per-event client multiplication for all transport types.

closes: #69284

---

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

• Yes: Claude Code (Claude Fable 5)

Generated-by: Claude Code following the guidelines

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[kacpermuda]

Looks good, one comment.

[gang-zh]

Hey @mobuchowski , can I get the PR merged to join the recent release?

[mobuchowski]

@gang-zh sorry, I wanted the tests to pass, took a long time and forgot to merge

[boring-cyborg]

Awesome work, congrats on your first merged pull request! You are invited to check our Issue Tracker for additional contributions.

[gang-zh]

Thank you! @mobuchowski this merged a few hours before 2.19.0rc1 was cut and didn't make the snapshot — any chance openlineage could be respun as 2.19.0rc2 in the current wave? It fixes a production scheduler CPU leak and unblocks re-enabling Datadog Jobs Monitoring for us.