Expose desiredNumUnboundedSourceSplits for Java and Python portable pipeline by Abacn · Pull Request #40345 · apache/beam

Abacn · 2026-09-29T19:50:26Z

Port desiredNumUnboundedSourceSplits currently honored by Dataflow non-portable pipeline submission to Java portable (UnboundedSource translated to SDF wrapper) and Python SDK

In Dataflow portable runner (Runner v2), unbounded GenerateSequence (backed by UnboundedSource) does not scale to reach target throughput at high rates because the SDF wrapper hardcodes the initial number of splits to 20 (DEFAULT_DESIRED_NUM_SPLITS = 20 in Java Read.java and _DEFAULT_DESIRED_NUM_SPLITS = 20 in Python unbounded_source.py), limiting read parallelism regardless of worker count.

In Java Dataflow Streaming Runner, initial splitting of UnboundedSources could be tuned via --desiredNumUnboundedSourceSplits in DataflowPipelineDebugOptions. Exposing this option in StreamingOptions and wiring it into the Java and Python SDF UnboundedSource wrappers allows portable pipelines to configure the desired number of initial splits and scale GenerateSequence (and other UnboundedSources) to high throughput targets.

…ipeline - Port desiredNumUnboundedSourceSplits currently honored by Dataflow non-portable pipeline submission to Java portable (UnboundedSource translated to SDF wrapper) and Python SDK

github-actions · 2026-09-30T15:18:49Z

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

Abacn · 2026-09-30T15:52:12Z

assign set of reviewers

github-actions · 2026-09-30T15:53:14Z

Assigning reviewers:

R: @tvalentyn for label python.
R: @kennknowles for label java.

This pull request likely touches a core component ("core" label). Please review with scrutiny.

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

tvalentyn

python bits lgtm

tvalentyn · 2026-09-30T16:23:55Z

looks like test suites fail on master too as per

Integration test triage — PR #40345

apache/beam @ 43e878d · 6 failing job(s) · triaged in 28.5s

12 failing test(s) across 6 suite(s) — 5 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 / beam_PreCommit_Python (Run Python PreCommit 3.10)	1	3	❌ failing (1h ago) — `gha_triage https://github.com/apache/beam/actions/runs/36735247726`
PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14)	1	3	❌ failing (1h ago) — `gha_triage https://github.com/apache/beam/actions/runs/36735247726`
PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10)	1	3	❌ failing (1h ago) — `gha_triage https://github.com/apache/beam/actions/runs/36732375686`
PreCommit Python / Python 3.10 Test Results () ⓘ published via the Checks API — not a job in this workflow	1	1	❌ failing (1h ago) — `gha_triage https://github.com/apache/beam/actions/runs/36735247726`
PreCommit Python / Test Results ⓘ published via the Checks API — not a job in this workflow	1	1	❌ failing (1h ago) — `gha_triage https://github.com/apache/beam/actions/runs/36735247726`
Build python source distribution and wheels / Build python cp310- wheels on auto for ubuntu-24.04	1	1	✅ passing (1h ago)

#	Suite	Verdict	Tests	Root cause
1	PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10) (+2)	LIKELY REAL BUG	6	AssertionError: Exception not raised
2	PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10) (+2)	LIKELY REAL BUG	3	AssertionError: False is not true : Did not find write call metric with status: already_exists
3	PreCommit Python / Python 3.10 Test Results ()	unknown	1	test_write_batch_error_surfaces_from_async_flush (apache_beam.io.gcp.bigtableio_test.TestWriteBigTable) failed — sdks/py
4	PreCommit Python / Test Results	unknown	1	classMethod (org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsubE2ETest) failed — examples/java/build/test-res
5	Build python source distribution and wheels / Build python cp310- wheels on auto for ubuntu-24.04	not a test failure	1	Process completed with exit code 1.

1. LIKELY REAL BUG — TestWriteBigTable (+5 more)

Suite: PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10), PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14), PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10)

Root cause: AssertionError: Exception not raised

Why this verdict: a test assertion failed

Affected tests:

TestWriteBigTable.test_write_batch_error_surfaces_from_async_flush (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10) — job log
TestWriteBigTable.test_write_batch_error_surfaces_from_buffered_rows (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10) — job log
TestWriteBigTable.test_write_batch_error_surfaces_from_async_flush (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14) — job log
TestWriteBigTable.test_write_batch_error_surfaces_from_buffered_rows (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14) — job log
TestWriteBigTable.test_write_batch_error_surfaces_from_async_flush (0s) — PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10) — job log
TestWriteBigTable.test_write_batch_error_surfaces_from_buffered_rows (0s) — PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10) — job log

Log excerpt

[gw5] linux -- Python 3.10.21 /runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310-cloud/py310-cloud/bin/python
self = <apache_beam.io.gcp.bigtableio_test.TestWriteBigTable testMethod=test_write_batch_error_surfaces_from_async_flush>
    def test_write_batch_error_surfaces_from_async_flush(self):
      write_fn = bigtableio._BigTableWriteFn(
          self._PROJECT_ID,
          self._INSTANCE_ID,
          self._TABLE_ID,
          flush_count=1,
          max_row_bytes=5242880)
      write_fn.table = self.table
      write_fn.start_bundle()
      direct_rows = [self.generate_row(i) for i in range(5)]
      with patch.object(Table,
                        'mutate_rows',
                        side_effect=Exception('batch RPC failed')):
        for direct_row in direct_rows:
          write_fn.process(direct_row)
>       with self.assertRaises(Exception):
E       AssertionError: Exception not raised
apache_beam/io/gcp/bigtableio_test.py:359: AssertionError

2. LIKELY REAL BUG — TestWriteBigTable (+2 more)

Suite: PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10), PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14), PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10)

Root cause: AssertionError: False is not true : Did not find write call metric with status: already_exists

Why this verdict: a test assertion failed

Affected tests:

TestWriteBigTable.test_write_metrics (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.10) — job log
TestWriteBigTable.test_write_metrics (0s) — PreCommit Python / beam_PreCommit_Python (Run Python PreCommit 3.14) — job log
TestWriteBigTable.test_write_metrics (0s) — PreCommit Python Coverage / beam_PreCommit_Python_Coverage (Run Python_Coverage PreCommit 3.10) — job log

Where (app code):

apache_beam/io/gcp/bigtableio_test.py:438 in verify_write_call_metric

Log excerpt

[gw5] linux -- Python 3.10.21 /runner/_work/beam/beam/sdks/python/test-suites/tox/py310/build/srcs/sdks/python/target/.tox-py310-cloud/py310-cloud/bin/python
self = <apache_beam.io.gcp.bigtableio_test.TestWriteBigTable testMethod=test_write_metrics>
    def test_write_metrics(self):
      MetricsEnvironment.process_wide_container().reset()
      write_fn = bigtableio._BigTableWriteFn(
          self._PROJECT_ID,
          self._INSTANCE_ID,
          self._TABLE_ID,
          flush_count=1000,
          max_row_bytes=5242880)
      write_fn.table = self.table
      write_fn.start_bundle()
      number_of_rows = 2
      error = Status()
      error.message = 'Entity already exists.'
      error.code = ALREADY_EXISTS
      success = Status()
      success.message = 'Success'
      success.code = OK
      rows_response = [error, success] * number_of_rows
      with patch.object(Table, 'mutate_rows', return_value=rows_response):
        direct_rows = [self.generate_row(i) for i in range(number_of_rows * 2)]
        for direct_row in direct_rows:
          write_fn.process(direct_row)
        try:
          write_fn.finish_bundle()
        except:  # pylint: disable=bare-except
          # Currently we fail the bundle when there are any failures.
          # TODO(https://github.com/apache/beam/issues/21396): remove after
          # bigtableio can selectively retry.

3. unknown — job-level failure

Suite: PreCommit Python / Python 3.10 Test Results ()

Root cause: test_write_batch_error_surfaces_from_async_flush (apache_beam.io.gcp.bigtableio_test.TestWriteBigTable) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 0s]

Why this verdict: published via the Checks API — not a job in this workflow

Affected tests:

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

Log excerpt

test_write_batch_error_surfaces_from_async_flush (apache_beam.io.gcp.bigtableio_test.TestWriteBigTable) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 0s]
test_write_batch_error_surfaces_from_buffered_rows (apache_beam.io.gcp.bigtableio_test.TestWriteBigTable) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 0s]
test_write_metrics (apache_beam.io.gcp.bigtableio_test.TestWriteBigTable) failed — sdks/python/test-suites/tox/py310/build/srcs/sdks/python/pytest_py310-cloudcoverage.xml [took 1s]

4. unknown — job-level failure

Suite: PreCommit Python / Test Results

Root cause: classMethod (org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsubE2ETest) failed — examples/java/build/test-results/test/TEST-org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsubE2ETest.xml [took 0s]

Why this verdict: published via the Checks API — not a job in this workflow

Affected tests:

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

Log excerpt

classMethod (org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsubE2ETest) failed — examples/java/build/test-results/test/TEST-org.apache.beam.examples.complete.kafkatopubsub.KafkaToPubsubE2ETest.xml [took 0s]

5. not a test failure — job-level failure

Suite: Build python source distribution and wheels / Build python cp310- wheels on auto for ubuntu-24.04

Root cause: Process completed with exit code 1.

Why this verdict: job failed outside of a test (policy check, setup, runner, or timeout)

Affected tests:

job-level failure.Build python source distribution and wheels / Build python cp310- wheels on auto for ubuntu-24.04 (0s) — Build python source distribution and wheels / Build python cp310- wheels on auto for ubuntu-24.04 — job log

Log excerpt

Process completed with exit code 1.

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

Abacn · 2026-10-01T15:23:21Z

GCP IO Direct different tests failed in two attempts, not related to this change

Abacn · 2026-10-01T16:48:52Z

Python PreCommit failed Bigtable tests not related to this change, fixed on master on #40364

Expose desiredNumUnboundedSourceSplits for Java and Python portable p…

43e878d

…ipeline - Port desiredNumUnboundedSourceSplits currently honored by Dataflow non-portable pipeline submission to Java portable (UnboundedSource translated to SDF wrapper) and Python SDK

github-actions Bot added python java io core labels Sep 29, 2026

Abacn marked this pull request as ready for review September 30, 2026 15:10

github-actions Bot requested review from kennknowles and tvalentyn September 30, 2026 15:53

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

tvalentyn approved these changes Sep 30, 2026

View reviewed changes

kennknowles approved these changes Oct 1, 2026

View reviewed changes

Abacn merged commit 355e8bd into apache:master Oct 1, 2026
142 of 158 checks passed

Abacn deleted the desired_split branch October 1, 2026 16:49

Navigation Menu

Repository navigation

Expose desiredNumUnboundedSourceSplits for Java and Python portable pipeline - #40345

Expose desiredNumUnboundedSourceSplits for Java and Python portable pipeline#40345
Abacn merged 1 commit into
apache:masterfrom
Abacn:desired_split

Abacn commented Sep 29, 2026 •

edited by tvalentyn

Loading

Uh oh!

github-actions Bot commented Sep 30, 2026

Uh oh!

Abacn commented Sep 30, 2026

Uh oh!

github-actions Bot commented Sep 30, 2026

Uh oh!

tvalentyn left a comment

Uh oh!

tvalentyn commented Sep 30, 2026

Uh oh!

Abacn commented Oct 1, 2026

Uh oh!

Abacn commented Oct 1, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Conversation

Abacn commented Sep 29, 2026 • edited by tvalentyn Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

github-actions Bot commented Sep 30, 2026

Uh oh!

Abacn commented Sep 30, 2026

Uh oh!

github-actions Bot commented Sep 30, 2026

Uh oh!

tvalentyn left a comment

Choose a reason for hiding this comment

Uh oh!

tvalentyn commented Sep 30, 2026

Integration test triage — PR #40345

1. LIKELY REAL BUG — TestWriteBigTable (+5 more)

2. LIKELY REAL BUG — TestWriteBigTable (+2 more)

3. unknown — job-level failure

4. unknown — job-level failure

5. not a test failure — job-level failure

Uh oh!

Abacn commented Oct 1, 2026

Uh oh!

Abacn commented Oct 1, 2026

Uh oh!

Uh oh!

Reviewers

Uh oh!

Assignees

Labels

Projects

Milestone

Development

Uh oh!

3 participants

Abacn commented Sep 29, 2026 •

edited by tvalentyn

Loading