[FLINK-40234][runtime] Exclude synchronous downstream processing from source idleness accounting

[slachiewicz]

What is the purpose of the change

Fixes FLINK-40234. Per-split idleness detection in SourceOperator measures time on a PausableRelativeClock that FLIP-471 pauses during backpressure and watermark-alignment pauses. Time the task thread spends inside the chained operators was still counted: a slow chained operator, or a window firing triggered by a watermark, that outlasts the idle timeout gets a split with fetched-but-unpolled records marked idle. The combined watermark then advances without that split and its records are dropped as late.

This PR pauses the input activity clock while downstream processing runs, the same treatment FLIP-471 gives to backpressure.

Brief change log

• PausableRelativeClock accepts any RelativeClock as base, so per-split clocks are now layered on the operator's main clock and one pause covers every split. The per-split backpressure listener registration goes away because the main clock, registered in SourceOperator#open, already provides it.
• New ActivityClockPausingDataOutput wraps the source's DataOutput and pauses the main clock around every downstream emit call (records, watermarks, statuses, latency markers, record attributes).
• The wrapper is installed only when a timestamp assigner or watermark generator asked the context for the input activity clock, which WatermarkStrategy#withIdleness does. Jobs without idleness keep the unwrapped output and an unchanged hot path.

Behaviour changes worth knowing

• Idleness is now measured in time during which the source could actually poll its splits. With chained operators taking a fraction f of the task thread, detecting a genuinely empty split next to a streaming one takes timeout / (1 - f) wall time instead of timeout. This trades slower idle detection for not dropping records, consistent with how backpressured time is already treated.
• Split clocks now also freeze while the operator waits for watermark alignment (WAITING_FOR_ALIGNMENT), which previously only froze the main clock. No split is polled in that mode.
• Time spent in other mailbox actions between polls (chained processing-time timers, checkpoint sync phase) is still counted. The operator cannot tell that time apart from a split being empty; making that exact needs the reader to report pending data per split, which is a connector-base API change and out of scope here. For the same reason the test from [FLINK-40234] Add unit test to demonstrate that busy subtask thread may cause splits to be marked as idle erroneously and cause dropped records #28816 is not adopted as written: it models the busy thread as time passing between two polls with no downstream call in flight.
• Per-record cost for jobs with idleness: one pause/unPause pair, measured at about 36 ns per element on an Apple M1 under JDK 17 (uncontended monitor plus two System.nanoTime reads). No cost for jobs without idleness.

Direction requested

This is the operator-level layer of the fix, and I would like the assignee's and reporter's view on whether it is the layer you want before polishing further:

1. Accept this trade-off: no false idleness from downstream busy time, at the price of slower idle detection under heavy downstream load (timeout / (1 - f)) and the per-record cost above for idleness users.
2. Go further and also exclude time between polls spent in other mailbox actions (timers, checkpoint sync phase). That covers the scenario modelled by [FLINK-40234] Add unit test to demonstrate that busy subtask thread may cause splits to be marked as idle erroneously and cause dropped records #28816, but it changes the semantics that five existing tests in SourceOperatorSplitWatermarkAlignmentTest encode (they advance time with records still queued in the mock reader and expect idleness), so those tests would need to be rewritten.
3. Fix it at the reader instead: SourceReaderBase knows when it holds fetched-but-unemitted records for a split and could keep that split's activity clock paused. That is exact per split and has no per-record cost, but needs a way to enumerate the split ids of a queued RecordsWithSplitIds and an internal hook from the reader to the split's clock, so it is a connector-base API change.

I lean towards 1 now and 3 as the follow-up. Happy to reshape this PR either way.

Verifying this change

• PausableRelativeClockTest#layeredClockFollowsBaseClockPauses covers clock layering.
• ActivityClockPausingDataOutputTest covers the decorator, including resuming the clock when the downstream call throws.
• SourceOperatorSplitWatermarkAlignmentTest#testSlowDownstreamRecordProcessingDoesNotMarkUnpolledSplitIdle and #testSlowDownstreamWatermarkProcessingDoesNotMarkUnpolledSplitIdle reproduce both scenarios from the ticket at the operator level. Both fail on master with Expecting value to be false but was true on the split's isIdle() after the busy stretch.
• The existing idleness, alignment and backpressure tests in that class pass unchanged.

Verified: mvn -pl flink-runtime test -Dtest=SourceOperatorSplitWatermarkAlignmentTest,ActivityClockPausingDataOutputTest,PausableRelativeClockTest → 21 tests, 0 failures; spotless:check and checkstyle:check clean (JDK 17, rebased on master c31f46757a4).

Does this pull request potentially affect one of the following parts

• Dependencies: no
• Public API: no, all touched classes are @Internal
• Serializers: no
• Performance-sensitive code paths: yes, see the per-record cost above; unchanged when idleness is not configured
• Deployment or recovery: no
• S3 file system connector: no

Documentation

• Does this pull request introduce a new feature? no

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

Generated-by: Claude Fable 5.1

[flinkbot]

CI report:

• 475625a Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[MartijnVisser]

@slachiewicz I think we're a bit premature with this PR, don't forget to also take a look at https://flink.apache.org/how-to-contribute/contribute-code/#1-create-jira-ticket-and-reach-consensus. You are opening a lot of PRs and we're happy with the contributions, but this is a hard problem that first needs consensus on how it should be resolved. Like this PR says, there is no consensus/working solution yet. Then we shouldn't open a PR. A problem like this can only be picked up when consensus has been reached and an assignee has been set, like the guide says