[Feature Request]: Flink runner: batch performance improvements #40607

What would you like to happen?

While running large Scio batch jobs on the Flink runner (Flink 1.20 and 2.2), profiling turned up several places where the runner spends a large share of the CPU on overhead rather than on user code, or spreads work unevenly across workers. This issue tracks a set of independent improvements, each in its own PR.

Findings

• Uneven source splits. The runner calls BoundedSource.split(estimatedSize / parallelism) once and keeps whatever comes back. Many sources don't honour the desired size. A BigQuery export, for example, returns one split per exported file: on one job, 714 files for 3750 readers left about 80% of the readers idle. Where there were enough splits, their sizes still varied a lot. On a 45 TiB Avro read at parallelism 3500, the busiest reader produced 1.38x the average records.
• Decode/re-encode in the batch sorter. Flink's sorter copies records with TypeSerializer#copy(DataInputView, DataOutputView) when it spills. Beam coders don't delimit their output, so CoderTypeSerializer does a full decode and re-encode, which was about half of the task CPU in a profiled job. Records sent to GroupByKey and Combine are also decoded on arrival and immediately re-encoded into the sorter, although only their key is needed.
• Metric updates from source readers. ReaderInvocationUtil pushes Beam metrics to Flink after every start()/advance() call. In a profiled job that was 37.6% of the read stage's CPU, more than reading the records.
• Windowing machinery in the global window. The final step of GroupByKey and Combine runs WindowDoFnOperator with ReduceFnRunner, which binds state, holds and timers for every key. That bookkeeping was 30-40% of the reduce phase, against about 6% for merging the accumulators.
• Random Reshuffle assignment. Flink's rebalance() starts each sender's round-robin at a random channel. When each sender emits only a few elements (file names, SDF restrictions), some receivers get nothing while others get several times the average.

Planned changes (all batch-only; streaming pipelines are unchanged)

• Re-split and balance bounded source splits, with static split assignment by default in batch (shared by the DataSet and DataStream translations) ([Flink] Re-split and balance bounded source splits in batch #40608)
• Length-prefix batch records so the sorter can copy them as bytes, and decode the values of grouping inputs lazily (--lengthPrefixBatchRecords, --lazyBatchGroupingInputs) ([Flink] Copy length-prefixed batch records and decode grouping inputs lazily #40609)
• Forward source reader metrics to Flink at most once per second (pipeline result metrics are unaffected) ([Flink] Throttle Flink metric updates from source readers #40610)
• A dedicated sorted operator for GroupByKey/Combine in the global window with the default trigger (--sortedBatchCombine) ([Flink] Add a dedicated operator for batch combines in the global window #40611)
• Deterministic round-robin start channels for Reshuffle ([Flink] Reshuffle batch inputs round-robin from deterministic start channels #40612)

The new options are enabled by default and can be turned off with --<option>=false. Static split assignment can be reverted to lazy with --sourceStaticSplitThresholdMb=0.

Issue Priority

Priority: 2 (default / most feature requests should be filed as P2)

Issue Components

• Component: Java SDK
• Component: Runner: Flink