Hey folks,
I had a pipe with sessionization restarts and then fail
after retries with this exception. The only thing I had done was to
increase the lateness by 12 hours ( to a day ) in this pipe and restart
from SP and it ran for 12 hours plus without issue. I cannot imagine that
increasing the lateness created this and the way I solved this was to
increase the lateness further. Could this be if there are TMs in the
cluster whose time is off ( as in not synchronized ) ?
2021-04-21 11:27:58
java.lang.UnsupportedOperationException: The end timestamp of an event-time
window cannot become earlier than the current watermark by merging. Current
watermark: 1618966593999 window: TimeWindow{start=1618878336107, end=
1618880140466}
at org.apache.flink.streaming.runtime.operators.windowing.
WindowOperator$2.merge(WindowOperator.java:339)
at org.apache.flink.streaming.runtime.operators.windowing.
WindowOperator$2.merge(WindowOperator.java:321)
at org.apache.flink.streaming.runtime.operators.windowing.
MergingWindowSet.addWindow(MergingWindowSet.java:209)
at org.apache.flink.streaming.runtime.operators.windowing.WindowOperator
.processElement(WindowOperator.java:319)
at org.apache.flink.streaming.runtime.tasks.
OneInputStreamTask$StreamTaskNetworkOutput.emitRecord(OneInputStreamTask
.java:191)
at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput
.processElement(StreamTaskNetworkInput.java:204)
at org.apache.flink.streaming.runtime.io.StreamTaskNetworkInput
.emitNext(StreamTaskNetworkInput.java:174)
at org.apache.flink.streaming.runtime.io.StreamOneInputProcessor
.processInput(StreamOneInputProcessor.java:65)
at org.apache.flink.streaming.runtime.tasks.StreamTask.processInput(
StreamTask.java:396)
at org.apache.flink.streaming.runtime.tasks.mailbox.MailboxProcessor
.runMailboxLoop(MailboxProcessor.java:191)
at org.apache.flink.streaming.runtime.tasks.StreamTask.runMailboxLoop(
StreamTask.java:617)
at org.apache.flink.streaming.runtime.tasks.StreamTask.invoke(StreamTask
.java:581)
at org.apache.flink.runtime.taskmanager.Task.doRun(Task.java:755)
at org.apache.flink.runtime.taskmanager.Task.run(Task.java:570)
at java.lang.Thread.run(Thread.java:748)
