[Bug][Flink]: Watermark may loss if split read too fast.

[zstraw]

What happened?

If there is an eventTime field defined in arctic table by 'create table t (wm as cast(op_time as timestamp(3)), watermark for wm as wm) like XXX' , watermark may not emitted.

Affects Versions

<=0.3.2

What engines are you seeing the problem on?

Flink

How to reproduce

1. Define a watermark by sql 'create table t (wm as cast(op_time as timestamp(3)), watermark for wm as wm) like XXX'.
2. Then group by tumble window, or other sql need watermark.
3. Insert into source table with one record.

You may find that there is no watermark in Flink web ui.

Relevant log output

No response

Anything else

The cause may be the case like:
The watermark in {@link WatermarkOutputMultiplexer.OutputState} which is associated with a split has been updated, but watermark may not been emitted for that when {@link WatermarkOutputMultiplexer#onPeriodicEmit} called, the outputState has been removed by {@link WatermarkOutputMultiplexer#unregisterOutput(String)} after split finished.

Code of Conduct

• I agree to follow this project's Code of Conduct