[BUG] PyFlink TypeError in WindowOperator when using allowed_lateness

[Umair Karel <um...@gmail.com>]

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