[FLINK-12872] WindowOperator may fail with UnsupportedOperationException when merging windows

• Type: Bug
• Status: Open
• Priority: Not a Priority
• Resolution: Unresolved
• Affects Version/s: 1.6.4, 1.7.2, 1.8.0
• Component/s: API / DataStream
• Labels: auto-deprioritized-major stale-minor
• Assignee: Unassigned
• Reporter: Piotr Nowojski
• Created: 17/Jun/19 11:42
• Updated: 16/Nov/21 15:19

Description

Reported by a user.

I have a job that uses processing time session window with inactivity gap of 60ms where I intermittently run into the following exception. I'm trying to figure out what happened here. Haven't been able to reproduce this scenario. Any thoughts?

java.lang.UnsupportedOperationException: The end timestamp of a processing-time window cannot become earlier than the current processing time by merging. Current processing time: 1560493731808 window: TimeWindow{start=1560493731654, end=1560493731778}
	at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator$2.merge(WindowOperator.java:325)
	at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator$2.merge(WindowOperator.java:311)
	at org.apache.flink.streaming.runtime.operators.windowing.MergingWindowSet.addWindow(MergingWindowSet.java:212)
	at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator.processElement(WindowOperator.java:311)
	at org.apache.flink.streaming.runtime.io.StreamInputProcessor.processInput(StreamInputProcessor.java:202)
	at org.apache.flink.streaming.runtime.tasks.OneInputStreamTask.run(OneInputStreamTask.java:105)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask.java:300)
	at org.apache.flink.runtime.taskmanager.Task.run(Task.java:711)
	at java.lang.Thread.run(Thread.java:745)

This is happening probably because System.currentTimeMillis() is not a monotonic function and WindowOperator accesses it at least twice: once when it creates a window and second time during performing the above mentioned check (that has failed). However I would guess there are more places like this, not only in WindowOperator.

The fix could be either to make sure that processing time is monotonic, or to access it only once per operator per record or to drop processing time in favour of ingestion time.

Issue Links

• is related to: FLINK-4421 Make clocks and time measurements monotonous