dheerajturaga opened a new pull request, #70690:
URL: https://github.com/apache/airflow/pull/70690
`CeleryExecutor` could stop scheduling indefinitely, with `/health`
returning 503
until the scheduler was restarted manually.
Publishing workloads forked a `ProcessPoolExecutor` from the scheduler,
which is
multi-threaded. `fork()` copies the state of every mutex but only the calling
thread, so a worker could inherit a lock held by a thread that was never
copied
and block forever acquiring it. The parent then waited on that worker
forever,
the scheduling loop stopped heartbeating, and the job never recovered on its
own.
`py-spy dump` of a wedged scheduler:
```
Thread (idle): "MainThread"
__enter__ (multiprocessing/synchronize.py:95) <--
SemLock.acquire(), never returns
get (multiprocessing/queues.py:100) <-- Queue.get():
"with self._rlock:"
_process_worker (concurrent/futures/process.py:242)
...
_send_workloads_to_celery (celery_executor.py:256)
_send_workloads (celery_executor.py:203)
trigger_tasks (base_executor.py:406)
heartbeat (base_executor.py:311)
_run_scheduler_loop (scheduler_job_runner.py:1693)
```
CPython already warns about this from the same code path — running the
existing
(quarantined) `test_send_workloads_to_celery_hang` on `main` emits
`This process is multi-threaded, use of fork() may lead to deadlocks in the
child`.
The existing `OPERATION_TIMEOUT` guard cannot catch it. That `timeout()`
lives
*inside* `send_workload_to_executor`, but the deadlock happens earlier, in
the
pool worker's bootstrap (`call_queue.get()`), before the target function is
entered. A `SIGALRM` also cannot interrupt a process blocked in a C-level
semaphore wait.
### Why remove the pool instead of changing the start method
The obvious fix is to pass a non-`fork` `mp_context`. Measured against a live
Redis broker (4 workers warm, `sync_parallelism` auto = 11), that is much
worse
than the status quo, because the pool is rebuilt on every publish:
| workloads | inline | fork pool | forkserver pool | persistent fork |
| --- | --- | --- | --- | --- |
| 5 | **3.2ms** | 51.2ms | 392.7ms | 3.6ms |
| 13 | **9.0ms** | 62.8ms | 2899ms | 6.4ms |
| 26 | **15.6ms** | 87.0ms | **24433ms** | 8.7ms |
| 128 | **72.4ms** | 186.4ms | 25981ms | 57.6ms |
| 512 | 364ms | 280ms | 42443ms | **152ms** |
Publishing a workload is a single short broker round-trip (~0.6ms), so the
pool
costs more to start than the sends it parallelizes. Inline beats the fork
pool at
every size up to 512 — and `[scheduler] max_tis_per_query` defaults to 16, so
real publish batches are far below where the pool could ever win. A per-batch
`forkserver`/`spawn` pool would turn an intermittent hang into a guaranteed
multi-second stall on every publish.
So publishing now happens in the scheduler process. That removes the hazard
rather than making the fork safe, and deletes the pool lifecycle, chunk
sizing,
and `BrokenProcessPool` handling that a persistent pool would have required.
`[celery] sync_parallelism` now applies only to fetching task state, which is
what it documents ("How many processes CeleryExecutor uses to sync task
state").
Deployments that set it to `1` to work around this hang can drop that
override.
### Bulk state fetch
`BulkStateFetcher._get_many_using_multiprocessing` has the identical fork
hazard
but genuinely parallelizes blocking network reads, so it keeps its pool and
asks
for a non-`fork` start method instead. It honours `[celery] mp_start_method`
then
`[core] mp_start_method` so an operator who has deliberately pinned one keeps
control; an explicit `fork` is respected with a warning rather than
overridden.
Unset, it selects `forkserver`, else `spawn`.
Note this path is only reached when `[celery] result_backend` is neither a
key-value store nor a database — Redis and DB backends route to
`_get_many_from_kv_backend` / `_get_many_from_db_backend` and never fork.
### Tests
`test_send_workloads_to_celery_hang` was quarantined *and* failing on
`main`: its
fixture passed a workload object in the tuple's `team_name` slot, so
`send_workload_to_executor` raised `AttributeError: 'MockWorkload' object
has no
attribute 'upper'`. In the forked pool that crash happened out of sight, so
the
test was quarantined rather than diagnosed. The fixture is fixed and the test
un-quarantined, since publishing no longer forks.
Added: publishing never starts a worker process (patched `os.fork` and
`ProcessPoolExecutor.__init__` both assert-fail), the state-fetch context
resolution chain including `[celery]` precedence over `[core]` and both
fallback
warnings, and that `BulkStateFetcher` passes a non-`fork` context. Each new
test
fails without the source change.
related: #8854 (a bulk publish API would remove this pool entirely)
related: #70163 (adjacent CeleryExecutor hang, different cause)
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes — Claude Code (Opus 5)
Generated-by: Claude Code (Opus 5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
