Fix SubprocessServer cache thread-safety and test isolation (#38501)

Author: shunping

.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
+}

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:

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):