This is an automated email from the ASF dual-hosted git repository. potiuk pushed a commit to branch backport-dadef4e9-v3-3-test in repository https://gitbox.apache.org/repos/asf/airflow.git
commit fe59b145c1045ca06697db31dce10e6804cfd6c7 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. (cherry picked from commit dadef4e91239f40775fc4a2615454d1ecbe42910) --- .../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 6fc7ba39a0a..8b7c5347f44 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(
