(airflow) branch main updated: Fix flaky LocalExecutor concurrent result refill test (#75030)

[henrychen]

This is an automated email from the ASF dual-hosted git repository.

henry3260 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 cffe6dfdcf1 Fix flaky LocalExecutor concurrent result refill test 
(#75030)
cffe6dfdcf1 is described below

commit cffe6dfdcf1a44040ce7bbe67ee3b48443df0b3c
Author: Shahar Epstein <60007259+shah...@users.noreply.github.com>
AuthorDate: Mon Oct 19 17:00:47 2026 +0300

    Fix flaky LocalExecutor concurrent result refill test (#75030)
    
    The test sent 16 KiB results for each ~0.9 KiB workload. When the dispatch
    loop blocked on a full activity pipe, the worker could fill the result pipe
    before it freed a page of the activity pipe, closing a two-pipe deadlock 
that
    drain-before-put cannot prevent at that ratio. It hangs intermittently, more
    often when pipes are smaller. Production results are a few hundred bytes,
    so the test now uses results smaller than a workload, which makes it
    deterministic with the fix, and dispatches enough workloads that it still
    deadlocks without it.
---
 airflow-core/tests/unit/executors/test_local_executor.py | 8 +++++---
 1 file changed, 5 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 0b4916c1390..d5004cffa4a 100644
--- a/airflow-core/tests/unit/executors/test_local_executor.py
+++ b/airflow-core/tests/unit/executors/test_local_executor.py
@@ -108,7 +108,7 @@ def _write_large_results_to_queue(result_queue, 
activity_queue, unread_messages,
 
 
 def _refill_worker(activity_queue, result_queue, barrier, count, payload_size):
-    """Stand-in worker: consume activity items and emit large results over 
real pipes."""
+    """Stand-in worker: consume activity items and emit results over real 
pipes."""
     payload = b"x" * payload_size
     barrier.wait()
     for _ in range(count):
@@ -491,7 +491,7 @@ class TestLocalExecutor:
         executor._dispatch_counts = {}
         mocker.patch.object(executor, "_check_workers", autospec=True)
 
-        count = 200
+        count = 400
         submitted = [_make_task_workload() for _ in range(count)]
         for workload in submitted:
             executor.executor_queues.setdefault(workload.type, 
{})[executor.get_workload_key(workload)] = (
@@ -499,9 +499,11 @@ class TestLocalExecutor:
             )
 
         barrier = ctx.Barrier(2)
+        # Keep results smaller than a workload, as in production: 
drain-before-put only guarantees progress
+        # while the worker can free a page of the activity pipe before it 
fills the result pipe.
         proc = ctx.Process(
             target=_refill_worker,
-            args=(executor.activity_queue, executor.result_queue, barrier, 
count, 16 * 1024),
+            args=(executor.activity_queue, executor.result_queue, barrier, 
count, 512),
         )
         proc.start()
         executor.workers = {proc.pid: proc}