slachiewicz opened a new pull request, #29152: URL: https://github.com/apache/flink/pull/29152
## 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 #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 #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? - [X] Yes (please specify the tool below) Generated-by: Claude Fable 5.1 -- 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]
