This is an automated email from the ASF dual-hosted git repository.
potiuk 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 aca7d5ab999 Fix flaky LocalExecutor drain test under Python 3.14
forkserver (#69538)
aca7d5ab999 is described below
commit aca7d5ab999ea298065c8334f0b91aaabaf0193c
Author: Jarek Potiuk <[email protected]>
AuthorDate: Tue Jul 7 13:45:51 2026 +0200
Fix flaky LocalExecutor drain test under Python 3.14 forkserver (#69538)
test_end_drains_result_queue_to_avoid_join_deadlock intermittently timed
out on Python 3.14, whose default multiprocessing start method on Linux is
now forkserver. The drain logic under test is start-method-agnostic, but a
forkserver worker is forked from a near-empty server and re-imports the
whole
airflow stack on every spawn; the throwaway worker therefore spends seconds
in
import airflow before writing a result, and end() waits through it,
occasionally
exceeding the test's execution_timeout.
Pin the test's worker to the fork start method so it inherits the
already-imported
parent, writes immediately, and reliably exercises the full-result_queue
scenario
the test guards -- as it did before 3.14 changed the default.
---
airflow-core/tests/unit/executors/test_local_executor.py | 12 +++++++++---
1 file changed, 9 insertions(+), 3 deletions(-)
diff --git a/airflow-core/tests/unit/executors/test_local_executor.py
b/airflow-core/tests/unit/executors/test_local_executor.py
index c4e3506f795..79987f9ab1c 100644
--- a/airflow-core/tests/unit/executors/test_local_executor.py
+++ b/airflow-core/tests/unit/executors/test_local_executor.py
@@ -347,12 +347,18 @@ class TestLocalExecutor:
@pytest.mark.execution_timeout(10)
def test_end_drains_result_queue_to_avoid_join_deadlock(self):
+ # Pin the worker to "fork": the drain logic under test is
start-method-agnostic, but under the
+ # "forkserver" default (Python 3.14+ on Linux) each spawned worker
re-imports the whole airflow
+ # stack before it can write a result, which intermittently exceeds the
execution_timeout and
+ # makes this test flaky. Forking inherits the already-imported parent,
so the worker writes
+ # immediately and reliably reproduces the full-result_queue scenario
this test guards.
+ ctx = multiprocessing.get_context("fork")
executor = LocalExecutor(parallelism=1)
- executor.activity_queue = multiprocessing.SimpleQueue()
- executor.result_queue = multiprocessing.SimpleQueue()
+ executor.activity_queue = ctx.SimpleQueue()
+ executor.result_queue = ctx.SimpleQueue()
result_count = 8
payload_size = 128 * 1024
- proc = multiprocessing.Process(
+ proc = ctx.Process(
target=_write_large_results_to_queue,
args=(executor.result_queue, result_count, payload_size),
)
