[Bug] Flink currentFetchEventTimeLag metric calculation is wrong when the datastream is completely blocked

[xiangyuf]

Search before asking

• [X] I searched in the issues and found nothing similar.

Paimon version

master(8ac121a)

Compute Engine

Flink

Minimal reproduce step

Check the Flink currentFetchEventTimeLag metric for a completely blocked datastream. It will not grow over time. But the currentEmitEventTimeLag will continue to grow.

What doesn't meet your expectations?

The currentFetchEventTimeLag metric should grow over time even if the datastream is completely blocked.

Anything else?

The lastSplitUpdateTime will not update for a completely blocked datastream.

In flink, the currentEmitEventTimeLag is calculated by idleTime or currentTime, so this metric can keep grow for a completely blocked datastream.

Are you willing to submit a PR?

• [X] I'm willing to submit a PR!