Hi all,
I would like to reach consensus on resolving FLINK-40234 [1] before
reviewing implementations. Although assigned to Roman, I investigated this
as the potential approaches have distinct trade-offs. Roman, please take
over the direction as you see fit.
Problem
Per-split idleness in SourceOperator measures inactivity via a
PausableRelativeClock (FLIP-471 / FLINK-35886), which pauses during
backpressure or watermark alignment but continues running while the task
thread executes chained operators. If downstream processing or
watermark-triggered window firings exceed the idle timeout, a split with
queued but unpolled records is incorrectly marked idle. Consequently, the
combined watermark advances, causing late, dropped records upon processing.
Keith's PR #28816 [2] demonstrates this failure, which I reproduced on
master.
Because SourceOperator cannot distinguish queued records from an empty
split—a state only the reader knows—we must choose where to implement the
fix.
Options
1. Operator level (pause during downstream processing): Wrap the source's
DataOutput to pause the activity clock during downstream
emitRecord/emitWatermark calls, layering per-split clocks onto the main
clock. Enabled only with withIdleness.
- Pros: Fixes both scenarios in FLINK-40234; no API changes; follows
FLIP-471 principles.
- Cons: Idleness is measured in source-active time (e.g., at 90%
downstream load, idle detection takes 10x wall time); adds ~36 ns/element
overhead for idleness users; does not exclude other mailbox actions
(timers, checkpoint sync).
2. Operator level (pause between polls): Pause clocks from MORE_AVAILABLE
until the next emitNext.
- Cons: Breaks semantics in existing tests (including FLINK-40093);
idle detection starves under light load because clocks only advance during
reader polling time. (I recommend against this.)
3. Reader level: SourceReaderBase tracks queued records per split and
pauses that split's clock until emission starts.
- Pros: Exact per-split tracking; no per-record overhead; preserves
empty-split semantics.
- Cons: Requires a connector-base API change (likely a FLIP) to
expose split IDs from RecordsWithSplitIds or the fetcher.
My recommendation: Implement Option 1 now as a bug fix (documenting the
semantic impact), followed by Option 3 as a long-term solution.
Prototype
Draft PR #29152 [3] implements Option 1 for reference (~120 lines of main
code plus tests). I will update or close it based on our decision.
Questions
a) Is Option 1's trade-off (preventing false idleness vs. slower idle
detection under downstream load) acceptable as a 2.4 bug fix, or should it
be opt-in?
b) Is anyone working on the reader-level signal (Option 3), or should we
create a follow-up ticket?
c) Roman, would you like to drive this discussion, or should I proceed?
[1] https://issues.apache.org/jira/browse/FLINK-40234
[2] https://github.com/apache/flink/pull/28816
[3] https://github.com/apache/flink/pull/29152
Best regards,
Sylwester Lachiewicz
