vlieven opened a new issue, #74544:
URL: https://github.com/apache/airflow/issues/74544
### Under which category would you file this issue?
Providers
### Apache Airflow version
3.3.1
### What happened and how to reproduce it?
The API server leaks processes forked by the OpenLineage listener until the
container is OOM-killed.
In Airflow 3, `patch_dag_run_state`
(`api_fastapi/core_api/services/public/dag_run.py`) calls `on_dag_run_success`
/ `on_dag_run_failed` from inside the API server when a DAG run is marked
success/failed through the UI or REST API. Manual task instance state changes
do the same through `_on_task_instance_manual_state_change`. The OpenLineage
listener hands these events to `OpenLineageListener.executor`, a
`ProcessPoolExecutor` created without an `mp_context`, so on Linux with Python
< 3.14 it **forks** the API server worker.
That causes two problems:
1. **Children deadlock.** The uvicorn worker is multithreaded (AnyIO worker
threads, DB pool, etc.), and forking a multithreaded process can leave the
child holding a lock that will never be released. Some children hang in
`futex_wait` and never send their event, so **lineage events are silently
lost**.
2. **Children are never cleaned up.** `before_stopping` is only called by
components such as the scheduler, never by API server workers, so the pool is
never shut down. Idle pool children block forever in `pipe_read`. They never
get EOF because they inherited the write end of their own call-queue pipe. When
gunicorn replaces a worker (`worker_refresh_interval`, timeout, crash), its
pool children are reparented to PID 1 and live forever. Each one is a full copy
of the worker (150-300 MiB PSS).
What we observed in one api-server pod, ~33 h after start:
```
master + 2 live workers ~580 MiB PSS
10 orphaned "gunicorn: worker" procs ~2,290 MiB PSS (ppid=1)
container memory.current 2,898 MiB of 3,272 MiB limit
```
Orphan states (`/proc/<pid>/wchan`): 5 × `pipe_read` (1 thread, idle pool
worker), 5 × `futex_wait_queue` (2 threads, deadlocked after fork).
`py-spy dump` of an orphan (the stack the child inherited from its parent at
fork time, followed by the pool worker loop):
```
Thread 184 (idle): "AnyIO worker thread"
_recv (multiprocessing/connection.py:395)
_recv_bytes (multiprocessing/connection.py:430)
recv_bytes (multiprocessing/connection.py:216)
get (multiprocessing/queues.py:103)
_process_worker (concurrent/futures/process.py:252)
run (multiprocessing/process.py:108)
_bootstrap (multiprocessing/process.py:314)
_launch (multiprocessing/popen_fork.py:71)
__init__ (multiprocessing/popen_fork.py:19)
_Popen (multiprocessing/context.py:282)
start (multiprocessing/process.py:121)
_spawn_process (concurrent/futures/process.py:799)
_launch_processes (concurrent/futures/process.py:789)
_start_executor_manager_thread (concurrent/futures/process.py:762)
submit (concurrent/futures/process.py:823)
submit_callable (openlineage/plugins/listener.py:971)
on_dag_run_success (openlineage/plugins/listener.py:893)
...pluggy...
patch_dag_run_state
(airflow/api_fastapi/core_api/services/public/dag_run.py:204)
patch_dag_run (airflow/api_fastapi/core_api/routes/public/dag_run.py:232)
run (anyio/_backends/_asyncio.py:1100)
```
Even without worker refresh, every API server worker that has handled a
manual state change keeps one extra forked process (~250 MiB) for its whole
lifetime. Worker refresh, which is meant to limit memory growth, turns that
into an unbounded leak.
### What you think should happen instead?
The listener should not fork from the API server. The process pool makes
sense in the scheduler (isolating event emission from the scheduler loop, see
#39235), but in the API server the work is a few HTTP calls. Possible fixes:
- Use a `ThreadPoolExecutor` when the listener runs outside the scheduler
(e.g. when the hook is invoked from the API server).
- Create the pool with a non-fork `mp_context` (`forkserver`/`spawn`) and
shut the pool down when an API server worker exits, so children cannot outlive
their worker. This avoids the deadlocks, but each child re-imports Airflow.
### Operating System
Debian (official `apache/airflow:slim-3.3.1-python3.12` image), Python
3.12.13
### Deployment
Other 3rd-party Helm chart
### Apache Airflow Provider(s)
openlineage
### Versions of Apache Airflow Providers
apache-airflow-providers-standard==1.11.1
apache-airflow-providers-cncf-kubernetes==10.17.0
apache-airflow-providers-openlineage==2.20.2
openlineage-python==1.53.0
### Official Helm Chart version
Not Applicable
### Kubernetes Version
1.35
### Helm Chart configuration
Not relevant
### Docker Image customizations
Not Applicable
### Anything else?
### Deployment details
- `airflow api-server` with `[api] server_type = gunicorn`, `[api] workers =
2`, `[api] worker_refresh_interval = 10800`
- gunicorn 26.2.0
- OpenLineage enabled with the `http` transport
- Kubernetes, api-server container memory limit ~3.2 GiB
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
