[Python SDK] Add support for named data streams to provide bundle isolation by scwhittle · Pull Request #40061 · apache/beam

scwhittle · 2026-09-08T12:05:04Z

This is advertised to the runner via a new NAMED_DATA_STREAMS protocol capability. The runner is then free to assign bundles to named data streams as it chooses to isolate bundle processing from each other. Instead of single data stream from the sdk, the sdk will create a data stream for each name. The benefit of doing so is that the multiplexing currently performed on data stream messages being received allows a slow bundle to fill up buffers and block the shared stream. With separate named streams, bundles on other data streams have separate grpc flow control from the blocked stream and are not affected.

This is related to issue #39001 and similar support was added for the Java sdk in #38863

Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:

Mention the appropriate issue in your description (for example: addresses #123), if applicable. This will automatically add a link to the pull request in the issue. If you would like the issue to automatically close on merging the pull request, comment fixes #<ISSUE NUMBER> instead.
Update CHANGES.md with noteworthy changes.
If this contribution is large, please file an Apache Individual Contributor License Agreement.

See the Contributor Guide for more tips on how to make review process smoother.

To check the build health, please visit https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md

GitHub Actions Tests Status (on master branch)

See CI.md for more information about GitHub Actions CI or the workflows README to see a list of phrases to trigger workflows.

github-actions · 2026-09-08T12:47:52Z

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

jrmccluskey

Broadly LGTM apart from linting and formatting findings

tvalentyn · 2026-09-12T05:02:37Z

Integration test triage — PR #40061

apache/beam @ b19c1eb · 15 failing job(s) · triaged in 20.7s

15 failing test(s) across 15 suite(s) — 3 distinct root cause(s).

Warning

At least one failure looks like a real defect, not a flake.

Failing suite	Jobs	Failures	On `master`
PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10)	1	1	✅ passing (2h ago)
PreCommit Python Formatter / beam_PreCommit_PythonFormatter (Run PythonFormatter PreCommit)	1	1	✅ passing (1h ago)
PreCommit Python Lint / Python 3.10 Test Results () ⓘ published via the Checks API — not a job in this workflow, and the PreCommit Python Lint workflow's own jobs all passed	1	1	✅ passing (1h ago)
Python tests / Python Unit Tests (macos-latest, 3.12, py312)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (windows-latest, 3.10, py310)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (macos-latest, 3.14, py314)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (windows-latest, 3.12, py312)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (windows-latest, 3.14, py314)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (macos-latest, 3.11, py311)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (windows-latest, 3.11, py311)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (macos-latest, 3.13, py313)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (windows-latest, 3.13, py313)	1	1	✅ passing (6h ago)
Python tests / Python Unit Tests (macos-latest, 3.10, py310)	1	1	✅ passing (6h ago)
PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.10)	1	1	✅ passing (2h ago)
PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.14)	1	1	✅ passing (2h ago)

#	Suite	Verdict	Tests	Root cause
1	PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10) (+12)	LIKELY REAL BUG	13	AssertionError: create_data_channel(api_service_descriptor {
2	PreCommit Python Formatter / beam_PreCommit_PythonFormatter (Run PythonFormatter PreCommit)	BUILD / LINT — must fix	1	file is not formatted (yapf --diff reported changes)
3	PreCommit Python Lint / Python 3.10 Test Results ()	unknown	1	test_named_data_streams_routing (apache_beam.runners.worker.bundle_processor_test.NamedDataStreamsTest) failed — sdks/py

1. LIKELY REAL BUG — NamedDataStreamsTest (+12 more)

Suite: PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10), Python tests / Python Unit Tests (macos-latest, 3.12, py312), Python tests / Python Unit Tests (windows-latest, 3.10, py310), Python tests / Python Unit Tests (macos-latest, 3.14, py314), Python tests / Python Unit Tests (windows-latest, 3.12, py312), Python tests / Python Unit Tests (windows-latest, 3.14, py314), Python tests / Python Unit Tests (macos-latest, 3.11, py311), Python tests / Python Unit Tests (windows-latest, 3.11, py311), Python tests / Python Unit Tests (macos-latest, 3.13, py313), Python tests / Python Unit Tests (windows-latest, 3.13, py313), Python tests / Python Unit Tests (macos-latest, 3.10, py310), PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.10), PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.14)

Root cause: AssertionError: create_data_channel(api_service_descriptor {

Why this verdict: a test assertion failed

Affected tests:

NamedDataStreamsTest.test_named_data_streams_routing (0s) — PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (macos-latest, 3.12, py312) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (windows-latest, 3.10, py310) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (macos-latest, 3.14, py314) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (windows-latest, 3.12, py312) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (windows-latest, 3.14, py314) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (macos-latest, 3.11, py311) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (windows-latest, 3.11, py311) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (macos-latest, 3.13, py313) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (windows-latest, 3.13, py313) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — Python tests / Python Unit Tests (macos-latest, 3.10, py310) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.10) — job log
NamedDataStreamsTest.test_named_data_streams_routing (0s) — PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.14) — job log

Log excerpt

[gw0] linux -- Python 3.10.21 /runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310-cloudcoverage/py310-cloudcoverage/bin/python
self = <apache_beam.runners.worker.bundle_processor_test.NamedDataStreamsTest testMethod=test_named_data_streams_routing>
    def test_named_data_streams_routing(self):
      descriptor = beam_fn_api_pb2.ProcessBundleDescriptor(id='descriptor_id')
      # Coders
      CODER_ID = 'coder'
      descriptor.coders[
          CODER_ID].spec.urn = common_urns.StandardCoders.Enum.BYTES.urn
      # PCollections
      PCOLLECTION_IN = 'pcoll_in'
      descriptor.pcollections[PCOLLECTION_IN].unique_name = PCOLLECTION_IN
      descriptor.pcollections[PCOLLECTION_IN].coder_id = CODER_ID
      PCOLLECTION_OUT = 'pcoll_out'
      descriptor.pcollections[PCOLLECTION_OUT].unique_name = PCOLLECTION_OUT
      descriptor.pcollections[PCOLLECTION_OUT].coder_id = CODER_ID
      # Source transform
      SOURCE_ID = 'source'
      source_transform = descriptor.transforms[SOURCE_ID]
      source_transform.spec.urn = bundle_processor.DATA_INPUT_URN
      source_port = beam_fn_api_pb2.RemoteGrpcPort(coder_id=CODER_ID)
      source_port.api_service_descriptor.url = 'localhost:123'
      source_transform.spec.payload = source_port.SerializeToString()
      source_transform.outputs['None'] = PCOLLECTION_IN
      # Sink transform
      SINK_ID = 'sink'
      sink_transform = descriptor.transforms[SINK_ID]
      sink_transform.spec.urn = bundle_processor.DATA_OUTPUT_URN
      sink_port = beam_fn_api_pb2.RemoteGrpcPort(coder_id=CODER_ID)
      sink_port.api_service_descriptor.url = 'localhost:123'
      sink_transform.spec.payload = sink_port.SerializeToString()

2. BUILD / LINT — must fix — formatting

Suite: PreCommit Python Formatter / beam_PreCommit_PythonFormatter (Run PythonFormatter PreCommit)

Root cause: file is not formatted (yapf --diff reported changes)

Why this verdict: formatter reported changes — run the formatter and commit

Affected tests:

formatting.bundle_processor_test.py (0s) — PreCommit Python Formatter / beam_PreCommit_PythonFormatter (Run PythonFormatter PreCommit) — job log

Log excerpt

--- apache_beam/runners/worker/bundle_processor_test.py	(original)
+++ apache_beam/runners/worker/bundle_processor_test.py	(reformatted)
@@ -793,10 +793,8 @@
     # Process on default stream
     processor.process_bundle('inst_1')
-    data_channel_factory.create_data_channel.assert_any_call(
-        source_port, None)
-    data_channel_factory.create_data_channel.assert_any_call(
-        sink_port, None)
+    data_channel_factory.create_data_channel.assert_any_call(source_port, None)
+    data_channel_factory.create_data_channel.assert_any_call(sink_port, None)
     mock_channel_default.output_stream.assert_called_once_with(
         'inst_1', SINK_ID)
Command exited with non-zero status 1
py3-yapf-check: exit 1 (180.94 seconds) /runner/_work/beam/beam/sdks/python/test-suites/tox/pycommon/build/srcs/sdks/python> time yapf --diff --parallel --recursive apache_beam pid=1927
1032.96user 13.01system 3:00.93elapsed 578%CPU (0avgtext+0avgdata 141500maxresident)k
py3-yapf-check: commands_post[0]> bash /runner/_work/beam/beam/sdks/python/test-suites/tox/pycommon/build/srcs/sdks/python/scripts/run_tox_cleanup.sh
34584inputs+32outputs (0major+553834minor)pagefaults 0swaps

3. unknown — job-level failure

Suite: PreCommit Python Lint / Python 3.10 Test Results ()

Root cause: test_named_data_streams_routing (apache_beam.runners.worker.bundle_processor_test.NamedDataStreamsTest) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 1s]

Why this verdict: published via the Checks API — not a job in this workflow, and the PreCommit Python Lint workflow's own jobs all passed

Affected tests:

job-level failure.PreCommit Python Lint / Python 3.10 Test Results () (0s) — PreCommit Python Lint / Python 3.10 Test Results () — job log

Log excerpt

test_named_data_streams_routing (apache_beam.runners.worker.bundle_processor_test.NamedDataStreamsTest) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 1s]

This tool can make mistakes. It is only meant to surface errors faster, and may not surface all of them accurately.

Generated by gha_triage

Initial support for named data streams for python sdk.

1220f56

github-actions Bot added python runners labels Sep 8, 2026

github-actions Bot added the Next Action: Reviewers label Sep 8, 2026

jrmccluskey reviewed Sep 9, 2026

View reviewed changes

jrmccluskey added Next Action: Author and removed Next Action: Reviewers labels Sep 9, 2026

fix format and lint::

ca39ed4

github-actions Bot added core Next Action: Reviewers and removed core Next Action: Author labels Sep 10, 2026

fix format and lint

b19c1eb

scwhittle force-pushed the python_named_streams branch from 0ee1c13 to b19c1eb Compare September 11, 2026 14:03

github-actions Bot added core and removed core labels Sep 11, 2026

Navigation Menu

Uh oh!

[Python SDK] Add support for named data streams to provide bundle isolation - #40061

[Python SDK] Add support for named data streams to provide bundle isolation#40061
scwhittle wants to merge 3 commits into
apache:masterfrom
scwhittle:python_named_streams

scwhittle commented Sep 8, 2026

Uh oh!

github-actions Bot commented Sep 8, 2026

Uh oh!

jrmccluskey left a comment

Uh oh!

tvalentyn commented Sep 12, 2026

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Uh oh!

Conversation

scwhittle commented Sep 8, 2026

GitHub Actions Tests Status (on master branch)

Uh oh!

github-actions Bot commented Sep 8, 2026

Uh oh!

jrmccluskey left a comment

Choose a reason for hiding this comment

Uh oh!

tvalentyn commented Sep 12, 2026

Integration test triage — PR #40061

1. LIKELY REAL BUG — NamedDataStreamsTest (+12 more)

2. BUILD / LINT — must fix — formatting

3. unknown — job-level failure

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants