This is an automated email from the ASF dual-hosted git repository.
vatsrahul1001 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 81976d301f2 Stop triggerer tests leaking runner processes (#76139)
81976d301f2 is described below
commit 81976d301f270710884d24fb066cb600e326ba8c
Author: Jarek Potiuk <[email protected]>
AuthorDate: Sat Oct 31 14:47:18 2026 +0100
Stop triggerer tests leaking runner processes (#76139)
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).
Generated-by: Claude Opus 5
---
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 888213d2753..18b0cccda82 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()
