This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch v3-4-test
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/v3-4-test by this push:
new 6593eadb474 [v3-4-test] Stop triggerer tests leaking runner processes
(#76139) (#76140)
6593eadb474 is described below
commit 6593eadb4742768c0125b8bfb59c97c85a1766e9
Author: github-actions[bot]
<41898282+github-actions[bot]@users.noreply.github.com>
AuthorDate: Sat Oct 31 19:20:03 2026 +0530
[v3-4-test] Stop triggerer tests leaking runner processes (#76139) (#76140)
Three triggerer tests ran a TriggerRunnerSupervisor until it handled its
events and left the forked runner running. Since the triggerer freezes the GC
before that fork, the supervisor is never collected and the runner never sees
EOF, so it outlives the test. On Python 3.12.16, ResourceTracker.__del__ waits
for the multiprocessing resource tracker at interpreter exit, and the leaked
runners keep its pipe open, so every 3.12 DB-core job hung after the Core tests
finished (#76135).
(cherry picked from commit 81976d301f270710884d24fb066cb600e326ba8c)
Generated-by: Claude Opus 5
Co-authored-by: Jarek Potiuk <[email protected]>
---
airflow-core/tests/unit/jobs/test_triggerer_job.py | 20 ++++++++++++++------
1 file changed, 14 insertions(+), 6 deletions(-)
diff --git a/airflow-core/tests/unit/jobs/test_triggerer_job.py
b/airflow-core/tests/unit/jobs/test_triggerer_job.py
index c1d09ac3b9d..73bbd8883d3 100644
--- a/airflow-core/tests/unit/jobs/test_triggerer_job.py
+++ b/airflow-core/tests/unit/jobs/test_triggerer_job.py
@@ -2666,6 +2666,17 @@ class
DummyTriggerRunnerSupervisor(TriggerRunnerSupervisor):
super().handle_events()
+def run_dummy_supervisor_until_events_handled(job: Job) -> None:
+ supervisor = DummyTriggerRunnerSupervisor.start(job=job, capacity=1,
logger=None)
+ try:
+ supervisor.run()
+ finally:
+ # run() returns once the events are handled, but the forked runner
keeps running. A leaked runner
+ # holds the multiprocessing resource tracker's pipe open, and on
Python 3.12.16 pytest then hangs
+ # at exit: ResourceTracker.__del__ waits for the tracker process,
which never sees EOF.
+ supervisor.kill(force=True)
+
+
@pytest.mark.asyncio
@pytest.mark.execution_timeout(20)
async def
test_trigger_can_call_variables_connections_and_xcoms_methods(session,
dag_maker):
@@ -2730,8 +2741,7 @@ async def
test_trigger_can_call_variables_connections_and_xcoms_methods(session,
session.add(job)
session.commit()
- supervisor = DummyTriggerRunnerSupervisor.start(job=job, capacity=1,
logger=None)
- supervisor.run()
+ run_dummy_supervisor_until_events_handled(job)
task_instance.refresh_from_db()
assert task_instance.state == TaskInstanceState.SCHEDULED
@@ -2834,8 +2844,7 @@ async def
test_trigger_can_fetch_trigger_dag_run_count_and_state_in_deferrable(s
session.add(job)
session.commit()
- supervisor = DummyTriggerRunnerSupervisor.start(job=job, capacity=1,
logger=None)
- supervisor.run()
+ run_dummy_supervisor_until_events_handled(job)
task_instance.refresh_from_db()
assert task_instance.state == TaskInstanceState.SCHEDULED
@@ -2935,8 +2944,7 @@ async def
test_trigger_can_fetch_dag_run_count_ti_count_in_deferrable(session, d
session.add(job)
session.commit()
- supervisor = DummyTriggerRunnerSupervisor.start(job=job, capacity=1,
logger=None)
- supervisor.run()
+ run_dummy_supervisor_until_events_handled(job)
parent_task.refresh_from_db()
task_instance.refresh_from_db()
