Hi Roman, Thanks for the reply. The evidence-based model makes sense to me, and I agree with defaulting to active. A missed idle transition stalls the watermark, which is visible and recoverable. A false idle transition drops records silently.
It also covers the case my prototype could not. Time spent in other mailbox actions (processing-time timers, the synchronous part of a checkpoint) cannot be told apart from an empty split with pausable clocks. In your model, the records are still buffered, so the split is simply never proven idle. For reference, I closed #29152 after Martijn's reminder about reaching consensus first, and I'm not planning to revive it. One open question: what is the plan for 2.4? If the FLIP targets a later release, FLINK-40234 stays a silent data-loss path until then. It would be good to hear from you or other committers whether an interim fix is wanted or whether the bug waits for the FLIP. A few points that may be worth covering in the FLIP: 1. Non-Kafka sources. "The broker reported no records" needs a connector-agnostic equivalent in connector-base. The file, Kinesis and Pulsar sources and custom SplitReaders need a way to report "fetch succeeded and was empty" per split. Without it, those sources could never prove idleness. 2. What "buffered anywhere" covers. Besides the fetcher and the SourceReaderBase element queue, it should include the RecordsWithSplitIds batch currently being emitted. Otherwise a smaller version of the same race remains. 3. User-visible behaviour change. With default-active, a split whose broker is unreachable never becomes idle, so the watermark stalls where today it would advance. That is the right trade-off, but withIdleness users should see it spelled out in the FLIP and the release notes. I'd also like to offer the two regression tests from #29152 as acceptance tests. They reproduce both scenarios from the ticket (slow downstream record processing and a slow watermark-triggered firing), and both fail on master. The core assertion, that a split with unpolled records is not marked idle, applies under any mechanism. The current harness drives time through the operator's test clock, though, so the setup would need adapting to your model. Once the direction is agreed, I can adapt them and contribute them standalone or against your branch, whichever you prefer. Happy to review the FLIP when it's up. Best regards, Sylwester On Fri, Sep 25, 2026 08:27 PM, Roman Khachatryan <[email protected]> wrote: > 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 > > >
