Hi Sylwester, Thanks for looking into this issue! We're actively working on a fix. But there seem to be multiple edge cases that are not covered today; and I believe we need to reconsider how we determine idleness.
The basic idea is: 1. Treat a split as active unless “proven” otherwise 2. Idleness “proof”: - Response from Kafka that there are no records - successful and recent , AND - No records buffered anywhere in Flink (SourceReader / Fetcher) 3. For each split, store the timestamps when it was seen empty, non-empty, processed records, records batch ID (fetched and processed) This gives the following advantages: 1. Defaults to Active => minimizes risks of data loss (due to bugs) 2. Covers Kafka connectivity issues and broker downtime by requiring a recent idle status from Broker 3. Covers Flink-side issues (such as back-pressure, business, etc.) by “passively” recording the last check time rather than “actively” pausing timers (pausing might be delayed itself or be missing due to a bug) 4. Simplicity => less error-prone, easy to reason about (easier than pauseable clocks) I'm planning to publish a FLIP later with a more detailed proposal. Regards, Roman On Thu, Sep 10, 2026 at 4:06 PM Sylwester Lachiewicz <[email protected]> wrote: > 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 >
