From 2a4b762363fe92f1d29ac873ffa79aa1343b1dd9 Mon Sep 17 00:00:00 2001
From: Shunping Huang <shunping@google.com>
Date: Thu, 14 May 2026 14:20:08 -0400
Subject: [PATCH] Fix SubprocessServer cache thread-safety and test isolation

---
 .../beam_PostCommit_Python_Versions.json              |  4 ++++
 sdks/python/apache_beam/transforms/external_test.py   |  8 +++++++-
 sdks/python/apache_beam/utils/subprocess_server.py    | 11 ++++++-----
 3 files changed, 17 insertions(+), 6 deletions(-)
 create mode 100644 .github/trigger_files/beam_PostCommit_Python_Versions.json

diff --git a/.github/trigger_files/beam_PostCommit_Python_Versions.json b/.github/trigger_files/beam_PostCommit_Python_Versions.json
new file mode 100644
index 000000000000..a975cd1cd104
--- /dev/null
+++ b/.github/trigger_files/beam_PostCommit_Python_Versions.json
@@ -0,0 +1,4 @@
+{
+  "comment": "Modify this file in a trivial way to cause this test suite to run",
+  "revision": 1
+}
diff --git a/sdks/python/apache_beam/transforms/external_test.py b/sdks/python/apache_beam/transforms/external_test.py
index 137c92861ed7..0af89367fcc2 100644
--- a/sdks/python/apache_beam/transforms/external_test.py
+++ b/sdks/python/apache_beam/transforms/external_test.py
@@ -799,7 +799,13 @@ def test_implicit_builder_with_constructor_method(self):
 
 class JavaJarExpansionServiceTest(unittest.TestCase):
   def setUp(self):
-    SubprocessServer._cache._live_owners = set()
+    # Temporarily override _live_owners with an empty set for this test,
+    # preventing contamination of the process-wide global cache and avoiding
+    # side effects on other tests.
+    patcher = mock.patch.object(
+        SubprocessServer._cache, '_live_owners', new=set())
+    patcher.start()
+    self.addCleanup(patcher.stop)
 
   def test_classpath(self):
     with tempfile.TemporaryDirectory() as temp_dir:
diff --git a/sdks/python/apache_beam/utils/subprocess_server.py b/sdks/python/apache_beam/utils/subprocess_server.py
index fed5ee591bcd..d21cb486b8f4 100644
--- a/sdks/python/apache_beam/utils/subprocess_server.py
+++ b/sdks/python/apache_beam/utils/subprocess_server.py
@@ -78,13 +78,14 @@ def __init__(self, constructor, destructor):
     self._counter = 0
 
   def _next_id(self):
-    with self._lock:
-      self._counter += 1
-      return self._counter
+    # Caller must hold self._lock.
+    self._counter += 1
+    return self._counter
 
   def register(self):
-    owner = self._next_id()
-    self._live_owners.add(owner)
+    with self._lock:
+      owner = self._next_id()
+      self._live_owners.add(owner)
     return owner
 
   def purge(self, owner):