[FLINK-27676] Output records from on_timer are behind the triggering watermark in PyFlink

• Type: Bug
• Status: Closed
• Priority: Major
• Resolution: Fixed
• Affects Version/s: 1.14.4, 1.15.0
• Fix Version/s: 1.14.5, 1.15.1, 1.16.0
• Component/s: API / Python
• Labels: pull-request-available
• Assignee: Juntao Hu
• Reporter: Juntao Hu
• Created: 18/May/22 07:44
• Updated: 19/May/22 12:24
• Resolved: 19/May/22 12:24

Description

Currently, when dealing with watermarks in AbstractPythonFunctionOperator, super.processWatermark(mark) is called, which advances watermark in timeServiceManager thus triggering timers and then emit current watermark. However, timer triggering is not synchronous in PyFlink (processTimer only put data into beam buffer), and when remote bundle is closed and output records produced by on_timer function finally arrive at Java side, they are already behind the triggering watermark.

Issue Links

• links to: GitHub Pull Request #19765