From baede6815da37c0b51b5c15183666aeaa6d9fba4 Mon Sep 17 00:00:00 2001 From: Bruno Volpato Date: Tue, 28 Jul 2026 00:44:20 -0400 Subject: [PATCH] Fix flaky AsyncWrapper reset_state test --- .../apache_beam/transforms/async_dofn_test.py | 15 +++++++++++++-- 1 file changed, 13 insertions(+), 2 deletions(-) diff --git a/sdks/python/apache_beam/transforms/async_dofn_test.py b/sdks/python/apache_beam/transforms/async_dofn_test.py index f9e07d57be9b..de5298f0a2c5 100644 --- a/sdks/python/apache_beam/transforms/async_dofn_test.py +++ b/sdks/python/apache_beam/transforms/async_dofn_test.py @@ -489,7 +489,7 @@ def add_item(i): self.assertEqual(bag_states['key' + str(i)].items, []) @staticmethod - def _run_reset_state_concurrent_teardown(use_asyncio): + def _run_reset_state_concurrent_teardown(use_asyncio, reset_started): dofn = BasicDofn(sleep_time=0.5) async_dofn = async_lib.AsyncWrapper(dofn, use_asyncio=use_asyncio) async_dofn.setup() @@ -502,15 +502,26 @@ def _run_reset_state_concurrent_teardown(use_asyncio): # Verify that calling reset_state() while background tasks are actively running # completes cleanly without causing lock-ordering deadlocks. + reset_started.set() async_lib.AsyncWrapper.reset_state() def test_reset_state_concurrent_teardown(self): # Verify concurrent teardown safety in a separate process to prevent any potential # regressions from freezing the main pytest process at exit. + reset_started = multiprocessing.Event() p = multiprocessing.Process( target=AsyncTest._run_reset_state_concurrent_teardown, - args=(self.use_asyncio, )) + args=(self.use_asyncio, reset_started)) p.start() + + # Python 3.14 uses forkserver by default on POSIX. Wait for child startup + # before measuring reset_state() so import and process startup time cannot + # be mistaken for a teardown deadlock. + if not reset_started.wait(timeout=30.0): + p.terminate() + p.join() + self.fail("child process did not reach reset_state() before timeout") + p.join(timeout=10.0) if p.is_alive():