pkuzmickas opened a new pull request, #39874: URL: https://github.com/apache/beam/pull/39874
## 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](https://github.com/apache/beam/blob/0b91ed1d432e6444511a4c5c766f9c175443a69e/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/FlinkSource.java#L104-L121). That enumerator [gives the next split to whichever reader requests it](https://github.com/apache/beam/blob/0b91ed1d432e6444511a4c5c766f9c175443a69e/runners/flink/src/main/java/org/apache/beam/runners/flink/translation/wrappers/streaming/io/source/LazyFlinkSourceSplitEnumerator.java#L108-L150). 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. A zero-byte estimate is valid and selects static assignment after opt-in. Negative and `Long.MAX_VALUE` estimates are treated as unknown and select lazy assignment. ## Implementation - Estimate size and split the source once on the enumerator's asynchronous worker. - Unify static and lazy assignment in one enumerator while preserving static assignment for unbounded sources. - 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. ## Validation - Focused `FlinkSourceSplitEnumeratorTest` suites passed for Flink 1.20, 2.0, and 2.2 before the branch was rebased and squashed; each suite ran 18 tests. - Flink Java formatting passed before the rebase. - The rebased commit passes `git diff --check`. A local rerun could not start because the available Java 25 runtime is unsupported by Beam's Gradle build; upstream CI will validate the exact commit. ------------------------ Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily: - [x] Mention the appropriate issue in the description. - [x] Update `CHANGES.md` with noteworthy changes. - [ ] Confirm the Apache ICLA before marking this draft ready. See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [making the review process smoother](https://github.com/apache/beam/blob/master/CONTRIBUTING.md#make-the-reviewers-job-easier). -- 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: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
