Retry starting SubprocessServer on failure

[shunping]

Wraps the subprocess startup and connection sequence in a retry loop with a limit of 3 attempts. This mitigates "address already in use" errors caused by race conditions during dynamic port allocation. Basically, we pick a random port but before using it to start a server it is occupied by other thread/worker.

An example failed test:
https://github.com/apache/beam/actions/runs/28321549831/job/83904243793

Traceback:

________________ CrossLanguageKinesisIOTest.test_kinesis_write _________________

self = <apache_beam.io.external.xlang_kinesisio_it_test.CrossLanguageKinesisIOTest testMethod=test_kinesis_write>

    @unittest.skipIf(
        TestPipeline().get_option('aws_kinesis_stream'),
        'Do not test on localstack when pipeline options were provided')
    def test_kinesis_write(self):
      # TODO: remove this test once
      # https://github.com/apache/beam/issues/20416 is resolved
>     self.run_kinesis_write()

apache_beam/io/external/xlang_kinesisio_it_test.py:97: 
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 
apache_beam/io/external/xlang_kinesisio_it_test.py:105: in run_kinesis_write
    with TestPipeline(options=PipelineOptions(self.pipeline_args)) as p:
apache_beam/pipeline.py:652: in __exit__
    self.result = self.run()
                  ^^^^^^^^^^
apache_beam/testing/test_pipeline.py:119: in run
    result = super().run(
apache_beam/pipeline.py:563: in run
    return self._run_internal(test_runner_api)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/pipeline.py:629: in _run_internal
    return self.runner.run_pipeline(self, self._options)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/runner.py:180: in run_pipeline
    return self.run_portable_pipeline(
apache_beam/runners/portability/portable_runner.py:375: in run_portable_pipeline
    job_service_handle = self.create_job_service(options)
                         ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/portability/portable_runner.py:289: in create_job_service
    return self.create_job_service_handle(server.start(), options)
                                          ^^^^^^^^^^^^^^
apache_beam/runners/portability/job_server.py:88: in start
    self._endpoint = self._job_server.start()
                     ^^^^^^^^^^^^^^^^^^^^^^^^
apache_beam/runners/portability/job_server.py:131: in start
    return self._server.start()
           ^^^^^^^^^^^^^^^^^^^^
_ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ _ 

self = <apache_beam.utils.subprocess_server.SubprocessServer object at 0x7881fc5f21d0>

    def start(self):
      try:
INFO     FlinkJarJobServer:subprocess_server.py:266 	at org.apache.beam.runners.flink.FlinkJobServerDriver.main(FlinkJobServerDriver.java:77)
INFO     FlinkJarJobServer:subprocess_server.py:266 Caused by: org.apache.beam.vendor.grpc.v1p69p0.io.netty.channel.unix.Errors$NativeIoException: bind(..) failed: Address already in use
INFO     FlinkJarJobServer:subprocess_server.py:266 [main] INFO org.apache.beam.runners.jobsubmission.JobServerDriver - ArtifactStagingServer stopped on localhost:36825
INFO     FlinkJarJobServer:subprocess_server.py:266 [main] INFO org.apache.beam.runners.jobsubmission.JobServerDriver - Expansion stopped on localhost:44161
ERROR    apache_beam.utils.subprocess_server:subprocess_server.py:226 Failed to start job service with ('java', '-jar', '-Dslf4j.provider=org.slf4j.simple.SimpleServiceProvider', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.streaming=error', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.runtime=error', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.flink.FlinkPipelineRunner=warn', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.fnexecution.control=warn', '/runner/_work/beam/beam/runners/flink/1.20/job-server/build/libs/beam-runners-flink-1.20-job-server-2.76.0-SNAPSHOT.jar', '--flink-master', '[auto]', '--artifacts-dir', '/tmp/beam-temp_3pmqphj/artifactsixp2vgsm', '--job-port', '36825', '--artifact-port', '0', '--expansion-port', '0')
ERROR    apache_beam.utils.subprocess_server:subprocess_server.py:240 Error bringing up service
Traceback (most recent call last):
  File "/runner/_work/beam/beam/sdks/python/apache_beam/utils/subprocess_server.py", line 227, in start
    raise RuntimeError(
RuntimeError: Service failed to start up with error 0
WARNING  apache_beam.utils.subprocess_server:subprocess_server.py:311 Really destroying service at localhost:36825 with cmd: ('java', '-jar', '-Dslf4j.provider=org.slf4j.simple.SimpleServiceProvider', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.streaming=error', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.runtime=error', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.flink.FlinkPipelineRunner=warn', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.fnexecution.control=warn', '/runner/_work/beam/beam/runners/flink/1.20/job-server/build/libs/beam-runners-flink-1.20-job-server-2.76.0-SNAPSHOT.jar', '--flink-master', '[auto]', '--artifacts-dir', '/tmp/beam-temp_3pmqphj/artifactsixp2vgsm', '--job-port', '36825', '--artifact-port', '0', '--expansion-port', '0')
Really destroying service at localhost:52571 with cmd: ('java', '-jar', '-Dslf4j.provider=org.slf4j.simple.SimpleServiceProvider', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.streaming=error', '-Dorg.slf4j.simpleLogger.log.org.apache.flink.runtime=error', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.flink.FlinkPipelineRunner=warn', '-Dorg.slf4j.simpleLogger.log.org.apache.beam.runners.fnexecution.control=warn', '/runner/_work/beam/beam/runners/flink/1.20/job-server/build/libs/beam-runners-flink-1.20-job-server-2.76.0-SNAPSHOT.jar', '--flink-master', '[auto]', '--artifacts-dir', '/tmp/beam-tempjp99srkd/artifactse8apq8m9', '--job-port', '52571', '--artifact-port', '0', '--expansion-port', '0')
WARNING  apache_beam.utils.subprocess_server:subprocess_server.py:311 Really destroying service at localhost:36579 with cmd: ['/opt/hostedtoolcache/Java_Temurin-Hotspot_jdk/21.0.11-10.0.LTS/x64/bin/java', '-jar', '/runner/_work/beam/beam/sdks/java/io/amazon-web-services2/expansion-service/build/libs/beam-sdks-java-io-amazon-web-services2-expansion-service-2.76.0-SNAPSHOT.jar', '36579', '--filesToStage=/runner/_work/beam/beam/sdks/java/io/amazon-web-services2/expansion-service/build/libs/beam-sdks-java-io-amazon-web-services2-expansion-service-2.76.0-SNAPSHOT.jar', '--alsoStartLoopbackWorker']

[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 introduces a robust retry mechanism for the SubprocessServer startup sequence. By wrapping the server initialization in a retry loop, the system can now gracefully recover from 'address already in use' errors that occur due to race conditions during dynamic port allocation, thereby increasing the stability of integration tests.

Highlights

• Retry Mechanism: Implemented a retry loop with a limit of 3 attempts for starting the SubprocessServer to handle transient port allocation conflicts.
• Error Handling: Improved error logging and added a 1-second delay between retry attempts to mitigate race conditions.
• Test Coverage: Updated unit tests to verify the retry logic and ensure proper cleanup of resources across all attempts.

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 retry logic (up to 3 attempts) with a 1-second delay when starting a subprocess server in subprocess_server.py, and updates the corresponding tests to verify this behavior. The reviewer noted that changing the bare except block to except Exception as e prevents catching BaseException (such as KeyboardInterrupt), which could lead to leaked subprocesses on interruption. They suggested adding a bare except block to handle cleanup in those cases.

Important

The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.

[github-actions]

Assigning reviewers:

R: @jrmccluskey for label python.

Note: If you would like to opt out of this review, comment assign to next reviewer.

Available commands:

• stop reviewer notifications - opt out of the automated review tooling
• remind me after tests pass - tag the comment author after tests pass
• waiting on author - shift the attention set back to the author (any comment or push by the author will return the attention set to the reviewers)

The PR bot will only process comments in the main thread (not review comments).