Fix SubprocessServer cache thread-safety and test isolation

[shunping]

I have seen the following intermittent log warnings in some test workflow logs, which seems to be double purge or some problems with server shutdown.

beam/sdks/python/apache_beam/utils/subprocess_server.py

Lines 95 to 96 in 4c4a2c1

	"Subprocess owner %s already purged. If this occurs during atexit "
	"shutdown, the subprocess was already cleaned up earlier.",

After some manual investigation (AI tooling not helpful here), however, I have revealed it is a global cache state that was being contaminated during tests.

Specifically, JavaJarExpansionServiceTest.setUp() was clearing the global _live_owners list. This caused any owners existed in the list to be lost from the registry, leading to the warning message when those owners get purged.

This PR fixes SubprocessServer cache thread safety and test isolation

• Use mock.patch.object in JavaJarExpansionServiceTest.setUp() to prevent contaminating the global _live_owners cache for other tests.
• Make _SharedCache.register() thread-safe.

[gemini-code-assist]

Summary of Changes

Hello, I'm Gemini Code Assist¹! I'm currently reviewing this pull request and will post my feedback shortly. In the meantime, here's a summary to help you and other reviewers quickly get up to speed!

This pull request addresses intermittent log warnings related to subprocess management in the Apache Beam Python SDK. By improving thread-safety in the SubprocessServer cache and enhancing test isolation for JavaJarExpansionServiceTest, the changes prevent global state contamination and ensure more reliable test execution. Additionally, diagnostic logging has been introduced to better monitor subprocess lifecycle events.

Highlights

• Thread-Safety: Updated SubprocessServer to ensure thread-safe registration of subprocess owners by wrapping the registration logic within a lock.
• Test Isolation: Improved test isolation in JavaJarExpansionServiceTest by using mock.patch.object to prevent contamination of the global _live_owners cache during test execution.
• Debugging: Added diagnostic buffers and logging to SubprocessServer to help trace intermittent subprocess purge warnings.

New Features

🧠 You can now enable Memory (public preview) to help Gemini Code Assist learn from your team's feedback. This makes future code reviews more consistent and personalized to your project's style. Click here to enable Memory in your admin console.

Using Gemini Code Assist

The full guide for Gemini Code Assist can be found on our documentation page, here are some quick tips.

Invoking Gemini

You can request assistance from Gemini at any point by creating a comment using either /gemini <command> or @gemini-code-assist <command>. Below is a summary of the supported commands on the current page.

Feature	Command	Description
Code Review	/gemini review	Performs a code review for the current pull request in its current state.
Pull Request Summary	/gemini summary	Provides a summary of the current pull request in its current state.
Comment	@gemini-code-assist	Responds in comments when explicitly tagged, both in pull request comments and review comments.
Help	/gemini help	Displays a list of available commands.

Customization

To customize the Gemini Code Assist for GitHub experience, repository maintainers can create a configuration file and/or provide a custom code review style guide (such as PEP-8 for Python) by creating and adding files to a .gemini/ folder in the base of the repository. Detailed instructions can be found here.

Limitations & Feedback

Gemini Code Assist may make mistakes. Please leave feedback on any instances where its feedback is incorrect or counterproductive. You can react with 👍 and 👎 on @gemini-code-assist comments. If you're interested in giving your feedback about your experience with Gemini Code Assist for GitHub and other Google products, sign up here.

Footnotes

1. Review the Privacy Notices, Generative AI Prohibited Use Policy, Terms of Service, and learn how to configure Gemini Code Assist in GitHub here. Gemini can make mistakes, so double check it and use code with caution. ↩

[gemini-code-assist]

Code Review

This pull request introduces debugging instrumentation to subprocess_server.py and associated tests, including internal buffers to track subprocess registration and purging, and enables global CLI logging in pytest.ini. It also refactors _next_id to rely on caller-side locking. The reviewer feedback highlights that the debugging buffers and hardcoded logic should be removed to prevent memory leaks and unnecessary overhead in production, and suggests reverting the logging changes to avoid excessive noise.

[gemini-code-assist]

Code Review

This pull request improves test isolation in JavaJarExpansionServiceTest by using mock.patch and refactors the locking mechanism in SubprocessServer to ensure atomicity during registration. Feedback highlights that removing the lock from _next_id makes the method thread-unsafe if called independently, suggesting the use of a re-entrant lock or inlining the logic. Additionally, there is a concern regarding the thread safety of patching global singletons when tests are run in parallel.

[shunping]

The failure of PreCommit Python PVR Flink is seen in the master branch as well, and it is irrelevant to the change here.

[shunping]

r: @derrickaw

[github-actions]

Checks are failing. Will not request review until checks are succeeding. If you'd like to override that behavior, comment assign set of reviewers

[github-actions]

Stopping reviewer notifications for this pull request: review requested by someone other than the bot, ceding control. If you'd like to restart, comment assign set of reviewers