[Flink] Throttle Flink metric updates from source readers

[jto]

Addresses #40607.

ReaderInvocationUtil pushed Beam metrics to Flink after every call to Reader#start and Reader#advance. Each push builds the metric results of all steps of the task and filters them down to one step. In a profiled batch job this took 37.6% of the CPU of the read stage, more than reading the records.

What changes

Metrics are pushed to Flink on the first call, at most once per second after that, and whenever the reader has no data. Metrics are still recorded on every call, and the metrics of the pipeline result, which come from the Flink accumulator, are not affected. Only the Flink metric views (e.g. the Flink UI) may lag by up to one second.

Testing

New ReaderInvocationUtilTest: Flink metrics are updated at most once per interval, when the reader has no data, and not at all when metrics are disabled.

[codecov]

Codecov Report

✅ All modified and coverable lines are covered by tests.
✅ Project coverage is 65.06%. Comparing base (76f6e6d) to head (4401f53).
⚠️ Report is 190 commits behind head on master.

Additional details and impacted files

@@             Coverage Diff              @@
##             master   #40610      +/-   ##
============================================
+ Coverage     56.16%   65.06%   +8.89%     
- Complexity     2288    13985   +11697     
============================================
  Files          1124     1729     +605     
  Lines        178336   108478   -69858     
  Branches       1489    11306    +9817     
============================================
- Hits         100170    70578   -29592     
+ Misses        75645    33696   -41949     
- Partials       2521     4204    +1683     

Flag	Coverage Δ
go	?
java	65.06% <100.00%> (-10.73%)	⬇️
python	?

Flags with carried forward coverage won't be shown. Click here to find out more.

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:

• ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.
• 📦 JS Bundle Analysis: Save yourself from yourself by tracking and limiting bundle sizes in JS merges.

[github-actions]

Assigning reviewers:

R: @Abacn added as fallback since no labels match configuration
Backup reviewer: chamikaramj (if the main reviewer is unresponsive, you can escalate to them by tagging their username)

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

[github-actions]

📋 Review Summary

This PR throttles Flink metric updates in ReaderInvocationUtil to at most once per second (plus the initial call and when start()/advance() returns false), addressing a major CPU bottleneck in batch source readers. The approach is clean and well-tested with a controllable clock, though the !dataAvailable condition can defeat throttling for UnboundedSource readers.

🔍 General Feedback

• ReaderInvocationUtil is shared across both bounded (SourceInputFormat, FlinkBoundedSourceReader) and unbounded (UnboundedSourceWrapper, FlinkUnboundedSourceReader) source implementations. While start()/advance() returning false signals EOF for a BoundedReader, for an UnboundedReader it only indicates that no record is immediately available.

---

💡 Tip: Comment /gemini review on this pull request to trigger a new Gemini code review.

[github-actions]

🟠 High: For UnboundedSource.UnboundedReader (used in FlinkUnboundedSourceReader and UnboundedSourceWrapper), start() / advance() returning false means no record is currently available rather than EOF.

Because both FlinkUnboundedSourceReader#nextReaderWithData and UnboundedSourceWrapper#run share a single ReaderInvocationUtil instance across all splits assigned to the subtask and poll them round-robin, if even one split is idle (or is an already-exhausted BoundedToUnboundedSourceAdapter split in FlinkUnboundedSourceReader), polling that split returns false before every record read from an active split. With || !dataAvailable, this bypasses the 1-second throttle and calls container.updateMetrics(stepName) on every single record (and every 50ms per split when completely idle).

Consider either restricting the !dataAvailable flush to BoundedSource.BoundedReader (e.g., passing reader instanceof BoundedSource.BoundedReader or checking it in maybeUpdateMetrics), or flushing final metrics explicitly when the source reader/operator closes.

[Abacn]

Thanks, this LGTM

Each push builds the metric results of all steps of the task and filters them down to one step.

Could be a follow up, any space optimize this insufficient way to push metrics?