This is an automated email from the ASF dual-hosted git repository.
tvalentyn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new dd463fa934d Fix flaky AsyncWrapper reset_state test on Python 3.14
(#39521)
dd463fa934d is described below
commit dd463fa934d1d8f3d3b459476d90ca55853e10e2
Author: Bruno Volpato <[email protected]>
AuthorDate: Tue Jul 28 06:58:28 2026 -0400
Fix flaky AsyncWrapper reset_state test on Python 3.14 (#39521)
---
sdks/python/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 f9e07d57be9..de5298f0a2c 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 @@ class AsyncTest(unittest.TestCase):
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 @@ class AsyncTest(unittest.TestCase):
# 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():
