[Flink] Select bounded-source split assignment by estimated size

[pkuzmickas]

Summary

Add an opt-in, size-based split-assignment strategy for bounded sources in the Flink DataStream runner.

The default remains lazy pull-based assignment. When users configure a positive threshold, the runner assigns sources estimated below that size per reader statically and keeps larger sources lazy. A negative value forces static assignment.

Fixes #39873.

Why

Moving production workloads from the Flink 1 DataSet runner to the Flink 2 DataStream runner caused large performance regressions for sources that emit inexpensive descriptors for expensive downstream work.

The DataStream runner currently sends all bounded sources to the lazy enumerator. That enumerator gives the next split to whichever reader requests it. This balances sources whose splits contain expensive I/O, but fast-starting readers can claim most descriptors before their peers start. A pointwise downstream edge then preserves the skew during the expensive work.

Static assignment fixes descriptor sources but can slow direct file readers, where dynamic work sharing compensates for different file sizes. The option therefore remains disabled by default and lets each pipeline choose a threshold.

Performance

These anonymized production measurements used matching input partitions and isolated outputs.

Workload	Source shape	Comparison	Result
A	Cheap descriptors followed by expensive work	Flink 2 static vs. Flink 1	~45% faster
B	Cheap descriptors followed by expensive work	Flink 2 static vs. Flink 1	~28% faster
C	Variable-size files read in the source	Threshold-selected lazy vs. forced static	~92 min vs. ~115 min (~20% faster)

The measured descriptor and file-reader examples were approximately 4 GiB and 8 GiB per reader, respectively. This supports a threshold near 6 GiB for those pipelines, but estimated bytes are only a proxy for split cost; Beam does not select a global threshold.

Configuration

lazySourceSplitAssignmentMinSizeMbPerReader (Python: lazy_source_split_assignment_min_size_mb_per_reader) controls bounded-source assignment:

• 0 (default): always use the existing lazy assignment.
• Positive: use static round-robin assignment below the threshold and lazy assignment at or above it.
• Negative: always use static assignment.

Implementation

• Estimate size and split the source once on the enumerator's asynchronous worker.
• Keep the static and lazy enumerators separate, and use a small size-based selector only when a positive threshold is configured.
• Queue reader requests while size-based selection is still running, then replay them to the selected enumerator.
• Preserve the existing Beam source bundle-size calculation; the new option changes assignment only.
• Persist the selected mode and a neutral list of pending splits in version 1 checkpoint state.
• Restore without estimating again; repartition pending static splits by split index when parallelism changes.
• Upgrade version 0 map checkpoints explicitly: bounded sources resume lazily, and unbounded sources resume statically.
• Document the Java and Python options.

Compatibility

Existing pipelines keep lazy bounded-source assignment because the option defaults to 0. Existing checkpoints retain their previous assignment behavior. Estimation failures continue to fail the job, matching the current runner.

---

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

• Mention the appropriate issue in the description.
• Update CHANGES.md with noteworthy changes.
• Confirm the Apache ICLA before marking this draft ready.

See the Contributor Guide for more tips on making the review process smoother.

[pkuzmickas]

Run Java PreCommit

[github-actions]

Assigning reviewers:

R: @shunping for label website.

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