You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by "scwhittle (via GitHub)" <gi...@apache.org> on 2026/09/08 12:05:07 UTC
[PR] [Python SDK] Add support for named data streams to provide bundle isolation [beam]
scwhittle opened a new pull request, #40061:
URL: https://github.com/apache/beam/pull/40061
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](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier).
To check the build health, please visit [https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md](https://github.com/apache/beam/blob/master/.test-infra/BUILD_STATUS.md)
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Go+tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI or the [workflows README](https://github.com/apache/beam/blob/master/.github/workflows/README.md) to see a list of phrases to trigger workflows.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python SDK] Add support for named data streams to provide bundle isolation [beam]
Posted by "tvalentyn (via GitHub)" <gi...@apache.org>.
tvalentyn commented on PR #40061:
URL: https://github.com/apache/beam/pull/40061#issuecomment-5643616635
## Integration test triage — PR #40061
_apache/beam @ b19c1ebd8 · 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)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Python_Coverage.yml?query=branch%3Amaster) |
| PreCommit Python Formatter / beam_PreCommit_PythonFormatter (Run PythonFormatter PreCommit) | 1 | 1 | [✅ passing (1h ago)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_PythonFormatter.yml?query=branch%3Amaster) |
| PreCommit Python Lint / Python 3.10 Test Results ()<br>ⓘ _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)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_PythonLint.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (macos-latest, 3.12, py312) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (windows-latest, 3.10, py310) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (macos-latest, 3.14, py314) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (windows-latest, 3.12, py312) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (windows-latest, 3.14, py314) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (macos-latest, 3.11, py311) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (windows-latest, 3.11, py311) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (macos-latest, 3.13, py313) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (windows-latest, 3.13, py313) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| Python tests / Python Unit Tests (macos-latest, 3.10, py310) | 1 | 1 | [✅ passing (6h ago)](https://github.com/apache/beam/actions/workflows/python_tests.yml?query=branch%3Amaster) |
| PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.10) | 1 | 1 | [✅ passing (2h ago)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Python_Runners.yml?query=branch%3Amaster) |
| PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.14) | 1 | 1 | [✅ passing (2h ago)](https://github.com/apache/beam/actions/workflows/beam_PreCommit_Python_Runners.yml?query=branch%3Amaster) |
| # | 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](https://github.com/apache/beam/actions/runs/34607852285/job/103290614744)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (macos-latest, 3.12, py312) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620161)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (windows-latest, 3.10, py310) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620276)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (macos-latest, 3.14, py314) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620291)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (windows-latest, 3.12, py312) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620336)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (windows-latest, 3.14, py314) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620416)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (macos-latest, 3.11, py311) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620489)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (windows-latest, 3.11, py311) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620499)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (macos-latest, 3.13, py313) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620535)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (windows-latest, 3.13, py313) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620540)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — Python tests / Python Unit Tests (macos-latest, 3.10, py310) — [job log](https://github.com/apache/beam/actions/runs/34607853926/job/103290620639)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.10) — [job log](https://github.com/apache/beam/actions/runs/34607852255/job/103290614937)
- `NamedDataStreamsTest.test_named_data_streams_routing` (0s) — PreCommit Python Runners / beam_PreCommit_Python_Runners (Run Python_Runners PreCommit 3.14) — [job log](https://github.com/apache/beam/actions/runs/34607852255/job/103290615112)
<details><summary>Log excerpt</summary>
```
[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()
```
</details>
### 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](https://github.com/apache/beam/actions/runs/34607852259/job/103290614581)
<details><summary>Log excerpt</summary>
```
--- 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
```
</details>
### 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](https://github.com/apache/beam/runs/103301682318)
<details><summary>Log excerpt</summary>
```
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]
```
</details>
_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_
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python SDK] Add support for named data streams to provide bundle isolation [beam]
Posted by "jrmccluskey (via GitHub)" <gi...@apache.org>.
jrmccluskey merged PR #40061:
URL: https://github.com/apache/beam/pull/40061
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] [Python SDK] Add support for named data streams to provide bundle isolation [beam]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #40061:
URL: https://github.com/apache/beam/pull/40061#issuecomment-5585373074
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).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@beam.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org