Hi, Thanks for investigating this and opening FLINK-41254. We tried a local change to LocalSlicingWindowAggOperator that calls windowBuffer.flush() before forwarding WatermarkStatus.IDLE. This fixed the initial buffered-record loss in our component tests. However, we found an additional scenario involving records received after the idle input resumes. For a 10-second tumbling window, the sequence is as follows. All timestamps below are in seconds:
* P1 flushes its existing buffer and becomes idle. P0 continues advancing, bringing the Global watermark to 120. * P1 resumes and receives a record with key=resumed, timestamp=121, and value=101, belonging to window [120,130). At this point, the record’s timestamp is ahead of the Global watermark. * P1 emits ACTIVE and watermark 118. Since its watermark remains behind the Global watermark, its channel is not yet included in the downstream minimum-watermark calculation. Meanwhile, Local keeps the record buffered because watermark 118 does not trigger its window. * P0 advances its watermark to 140, allowing Global to pass the window’s acceptance boundary before receiving P1’s partial aggregate. * When P1 becomes idle again, Local flushes the record. By then, the Global watermark has reached 156, and the partial aggregate is dropped as late. The expected COUNT=1, SUM=101 result for this key is therefore missing. We also reproduced this recovery scenario with the unmodified Local operator by invoking prepareSnapshotPreBarrier()before the first idle transition to clear the earlier buffer. Without that early flush, an older buffered window can trigger a whole-buffer flush on watermark 118, incidentally forwarding the resumed record in time and masking this path. The scenario remained reproducible with the FLINK-40504 and FLINK-40475 changes applied. These results come from deterministic component tests; we have not yet validated the candidate change in a complete Kafka/Flink deployment. Could this recovery scenario also be considered in the upcoming PR? It suggests that flushing before forwarding IDLE addresses the initial loss, but additional handling may be needed for records buffered after reactivation while the channel’s watermark is still catching up. Best regards, Yuquan Chen
