bingqin2 opened a new pull request, #76843:
URL: https://github.com/apache/airflow/pull/76843
closes: #70117
related: #70163
When the exit status of a supervised process has already been collected
elsewhere, `psutil.Process.wait(timeout=0)` returns `None` instead of an exit
code, on every call. One way this happens is `SIGCHLD` being ignored
(`SIG_IGN`) in the supervisor process. The kernel then reaps the child as soon
as it exits, and the setting is inherited across `fork` and `exec`.
On main, that `None` makes the supervisor wait forever:
- `WatchedSubprocess._check_subprocess_exit` stored the `None` as
`_exit_code` and reset `_process_exit_monotonic` on every call.
- `_monitor_subprocess` loops `while self._exit_code is None or
self._open_sockets`, so it never ended.
- The `socket_cleanup_timeout` check requires an exit code, so it never ran.
- The supervisor kept sending heartbeats for a task process that was gone.
The issue reports 26–30 such supervisors per Celery worker, at about 340 MB
each.
With this change:
- A `None` from `wait()` is recorded once, as an exit whose status was lost.
The exit code becomes `-1`, which is what `kill()` already records for a
process that is already gone, and the task log gets a warning. The exit time is
recorded only once, so `socket_cleanup_timeout` still applies. Normal exits are
unchanged; #70163 broke this for them.
- When the status was lost, `ActivitySubprocess.final_state` uses the state
the task reported. Without this, the unknown exit code would turn a reported
success into a failure. If the task reported no state, it fails or retries, as
for any nonzero exit. A known nonzero exit code still overrides a reported
state, as before.
The fix is in `WatchedSubprocess`, so it also covers the Dag processor and
triggerer subprocesses. I found no `SIGCHLD` handler or `waitpid(-1)` call in
Celery 5 or billiard 4, so what collects the status in the reported deployment
is unconfirmed. The fix does not depend on it.
**Tests** (`task-sdk/tests/task_sdk/execution_time/test_supervisor.py`)
- `test_exit_status_collected_elsewhere`: a real task process under
`SIG_IGN`, in two cases:
- Nothing reported: the task finishes FAILED.
- Success reported: only `succeed()` is called, with no `finish(FAILED)`
after it.
On main both cases hit the 30 s timeout, because the supervisor waits
forever.
- `test_exit_status_collected_elsewhere_is_recorded_once`: later checks do
not reset the exit time.
- `test_lost_exit_status_uses_reported_state`: the `final_state` cases,
including a known nonzero exit code still overriding a reported success.
**Testing**
- `task-sdk`, `test_supervisor.py`, on main da7dab32a8: 382 passed, 3
skipped.
- 4 tests fail the same way on main in my local environment. They are
log-content assertions that see an extra "Secrets backends loaded for worker"
line: `test_reading_from_pipes`, both `test_run_simple_dag` cases and
`test_supervise_handles_deferred_task`.
- All 8 new tests fail on main. The known-exit-code case fails only
because the new attribute does not exist there; its expected result is main's
current behavior.
- `airflow-core`: `tests/unit/dag_processing/test_processor.py` and
`tests/unit/jobs/test_triggerer_job.py`, 349 passed, 1 skipped.
- ruff and mypy on the changed files.
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Claude Opus 5.5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions).
I reviewed and understand all changes; the tests were run locally as listed
above.
---
🤖 Generated with [Claude Code](https://claude.com/claude-code)
--
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]
