(airflow) branch v3-4-test updated: Fix flaky LocalExecutor concurrent result refill test (#75030) (#76117)

[rahulvats.apache.org via commits]

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

commit 1c8d16792213e2cf5998cda1d1b016d0e697af69
Author: Rahul Vats <43964496+vatsrahul1...@users.noreply.github.com>
AuthorDate: Sat Oct 31 12:41:44 2026 +0530

    Fix flaky LocalExecutor concurrent result refill test (#75030) (#76117)
    
    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.
    
    (cherry picked from commit cffe6dfdcf1a44040ce7bbe67ee3b48443df0b3c)
    
    Co-authored-by: Shahar Epstein <60007259+shah...@users.noreply.github.com>
---
 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 08d95cf370c..92572b9844e 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}