Parent issue: #12340
Context:
When a Flink source (e.g. Kafka) stops producing records, Flink requires WatermarkStatus.IDLE to be emitted so downstream operators (e.g. window aggregations) can advance their own watermarks independently. Currently the Gluten native source pipeline does not detect idleness or emit idle status.
Scope:
- Implement native-side idle detection using
SystemProcessingTimeScheduler timers
- Add
WatermarkIdleTracker (C++) with configurable idle timeout and watermark interval
- Integrate idle detection into
WatermarkSource / WatermarkGenerator and WatermarkAssigner
- Add
WatermarkStatus.IDLE / ACTIVE emission path through NativeCallbackBridge → Java source function → SourceContext.markAsTemporarilyIdle()
- Add
shouldCallNoMoreSplits flag to GlutenSourceFunction (default true, set false for unbounded streaming to keep task alive)
- Preserve the flag through
OffloadedJobGraphGenerator pass-through
- End-to-end test with embedded Kafka broker + Flink MiniCluster verifying idle detection
Status: Implementation complete. See PR for details.
Parent issue: #12340
Context:
When a Flink source (e.g. Kafka) stops producing records, Flink requires
WatermarkStatus.IDLEto be emitted so downstream operators (e.g. window aggregations) can advance their own watermarks independently. Currently the Gluten native source pipeline does not detect idleness or emit idle status.Scope:
SystemProcessingTimeSchedulertimersWatermarkIdleTracker(C++) with configurable idle timeout and watermark intervalWatermarkSource/WatermarkGeneratorandWatermarkAssignerWatermarkStatus.IDLE/ACTIVEemission path throughNativeCallbackBridge→ Java source function →SourceContext.markAsTemporarilyIdle()shouldCallNoMoreSplitsflag toGlutenSourceFunction(defaulttrue, setfalsefor unbounded streaming to keep task alive)OffloadedJobGraphGeneratorpass-throughStatus: Implementation complete. See PR for details.