Hi Flink Community, I am writing to report a bug I encountered in the PyFlink DataStream API (tested on Flink 1.20.0, and the code appears unchanged on the master branch as of today). I was unable to create an account on Jira to log this formally, so I am sharing the details here.
*The Bug* When configuring .allowed_lateness() on a WindowedStream using a PyFlink Time object (e.g., Time.seconds(50)), the TaskManager Python worker crashes during the window evaluation phase. *Stack Trace* *File "/usr/local/lib/python3.10/dist-packages/pyflink/fn_execution/datastream/window/window_operator.py", line 514, in cleanup_time time = window.max_timestamp() + self.allowed_latenessTypeError: unsupported operand type(s) for +: 'int' and 'Time'* *Steps to Reproduce* 1. Set up a PyFlink DataStream with watermarks. 2. Define a window and call .allowed_lateness(Time.seconds(50)). *windowed_stream = ds \ .key_by(lambda x: x[0]) \ .window(TumblingEventTimeWindows.of(Time.minutes(1))) \ .allowed_lateness(Time.seconds(50)) \ .process(MyProcessWindowFunction(), Types.ROW(...))* 3. Push late events to trigger the late data pathway. *Root Cause Analysis* The official type hint for WindowOperator.__init__ indicates that allowed_lateness should be an int. However, at runtime, the API passes the raw pyflink.common.time.Time object. During JobGraph translation, PyFlink uses cloudpickle to serialize the WindowOperationDescriptor directly into a byte array, which is passed through Java and unpacked by the Python worker. Because it bypasses Java entirely, the TaskManager worker receives a Time object instead of an integer. When cleanup_time() evaluates window.max_timestamp() + self.allowed_lateness, it attempts to add an integer to a Time object, resulting in the crash. *Suggested Fix* The fix is a simple adjustment to the Python SDK to safely extract the milliseconds. In *flink-python/pyflink/fn_execution/datastream/window/window_operator.py* inside *cleanup_time(self, window):* *Current:* *def cleanup_time(self, window) -> int: if self.window_assigner.is_event_time(): time = window.max_timestamp() + self.allowed_lateness* *Proposed:* *def cleanup_time(self, window) -> int: if self.window_assigner.is_event_time(): lateness_ms = self.allowed_lateness.to_milliseconds() if hasattr(self.allowed_lateness, "to_milliseconds") else self.allowed_lateness time = window.max_timestamp() + lateness_ms* Could someone with Jira access please open a ticket for this, or let me know if you need any additional information? -- Regards, Umair Karel Linkedin: https://www.linkedin.com/in/umairkarel/ Github: https://github.com/umairkarel
