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 dadef4e9123 Fix flaky bundle version lock concurrency tests (#72572)
dadef4e9123 is described below
commit dadef4e91239f40775fc4a2615454d1ecbe42910
Author: Jyun-An Chen <[email protected]>
AuthorDate: Wed Sep 9 09:31:42 2026 +0800
Fix flaky bundle version lock concurrency tests (#72572)
The bundle version lock concurrency tests assumed a worker thread had
already acquired the lock after a fixed 0.1s sleep. That assumption only
holds on an idle machine: under CI-like contention the thread can take
longer than that just to be scheduled, so the tests failed for reasons
unrelated to the code under test.
A failure was also worse than it needed to be. The lock object was
assigned inside the worker thread, so a test that ran too early raised
AttributeError instead of pointing at the timing; and the stop flag was
only set on the success path while join() had no timeout, so a failed
assertion left the worker looping forever and hung the job instead of
failing it.
---
.../tests/unit/dag_processing/bundles/test_base.py | 75 +++++++++++++---------
1 file changed, 43 insertions(+), 32 deletions(-)
diff --git a/airflow-core/tests/unit/dag_processing/bundles/test_base.py
b/airflow-core/tests/unit/dag_processing/bundles/test_base.py
index f092f3e00e7..83ccd645066 100644
--- a/airflow-core/tests/unit/dag_processing/bundles/test_base.py
+++ b/airflow-core/tests/unit/dag_processing/bundles/test_base.py
@@ -21,7 +21,6 @@ import fcntl
import logging
import tempfile
import threading
-import time
from datetime import timedelta
from pathlib import Path
from unittest.mock import call, patch
@@ -138,61 +137,73 @@ def test_lock_exception_handling():
assert acquired
+LOCK_WAIT_TIMEOUT = 10
+
+
class LockTestHelper:
def __init__(self, num, **kwargs):
super().__init__(**kwargs)
self.num = num
- self.stop = None
- self.did_lock = None
- self.locker: BundleVersionLock
-
- def lock_the_file(self):
+ self.stop = threading.Event()
+ self.did_lock = threading.Event()
self.locker = BundleVersionLock(
bundle_name="abc",
bundle_version="this",
)
+
+ def lock_the_file(self):
with self.locker:
- self.did_lock = True
+ self.did_lock.set()
idx = 0
- while not self.stop:
+ while not self.stop.wait(0.2):
idx += 1
- time.sleep(0.2)
log.info("sleeping: idx=%s num=%s", idx, self.num)
log.info("exit")
+ def start(self):
+ thread = threading.Thread(target=self.lock_the_file)
+ thread.start()
+ return thread
+
+ def wait_until_locked(self):
+ assert self.did_lock.wait(LOCK_WAIT_TIMEOUT), f"helper {self.num}
never acquired the lock"
+
class TestBundleVersionLock:
def test_that_shared_lock_doesnt_block_shared_lock(self):
"""Verify that two things can lock file at same time."""
lth1 = LockTestHelper(1)
- t1 = threading.Thread(target=lth1.lock_the_file)
lth2 = LockTestHelper(2)
- t2 = threading.Thread(target=lth2.lock_the_file)
- t1.start()
- time.sleep(0.1)
- assert lth1.did_lock is True
- t2.start()
- time.sleep(0.1)
- assert lth2.did_lock is True
- lth1.stop = True
- lth2.stop = True
- t1.join()
- t2.join()
+ t1 = lth1.start()
+ t2 = None
+ try:
+ lth1.wait_until_locked()
+ t2 = lth2.start()
+ lth2.wait_until_locked()
+ finally:
+ lth1.stop.set()
+ lth2.stop.set()
+ t1.join(LOCK_WAIT_TIMEOUT)
+ if t2:
+ t2.join(LOCK_WAIT_TIMEOUT)
+ assert not t1.is_alive()
+ assert not t2.is_alive()
def test_that_shared_lock_blocks_ex_lock(self):
"""Test that exclusive lock is impossible when in bundle lock
context."""
lth1 = LockTestHelper(1)
- t1 = threading.Thread(target=lth1.lock_the_file)
- t1.start()
- time.sleep(0.1)
- assert lth1.did_lock is True
- with open(lth1.locker.lock_file_path, "a") as f:
- fcntl.flock(f, fcntl.LOCK_SH)
- fcntl.flock(f, fcntl.LOCK_UN)
- with pytest.raises(BlockingIOError): # <-- this is the important
part
- fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
- lth1.stop = True
- t1.join()
+ t1 = lth1.start()
+ try:
+ lth1.wait_until_locked()
+ with open(lth1.locker.lock_file_path, "a") as f:
+ fcntl.flock(f, fcntl.LOCK_SH)
+ fcntl.flock(f, fcntl.LOCK_UN)
+ with pytest.raises(BlockingIOError): # <-- this is the
important part
+ fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
+ finally:
+ lth1.stop.set()
+ t1.join(LOCK_WAIT_TIMEOUT)
+ assert not t1.is_alive()
def test_that_no_version_is_noop(self):
with BundleVersionLock(
