[flink][hotfix] Wait for consumer reset before job close

[yunfengzhou-hub]

Purpose

Linked issue: #4442

This PR fixes the bug that Consumers might not have been reset yet when the streaming job is closed in test cases. This is because the resetting logic happens in MonitorFunction#notifyCheckpointComplete, and Flink does not guarantee the invocation of this method when the job is canceled or stopped.

This bug happens rarely in the current test cases because Flink's legacy source function runs independently in a separated thread. When we use FLIP-27 source for MonitorFunction and checkpoint events may pile up in Flink's Mailbox, this bug will be triggered more frequently.

Tests

Existing test cases are used to verify this change.

API and Format

This change does not affect API or storage format.

Documentation

This change does not affect introduce a new feature.

[wwj6591812]

Hi, @yunfengzhou-hub .
You means we will use FLIP-27 source to rewrite MonitorFunction?
If yes, I have not see this change in https://github.com/apache/paimon/pull/4567/files.

[yunfengzhou-hub]

Hi, @yunfengzhou-hub . You means we will use FLIP-27 source to rewrite MonitorFunction? If yes, I have not see this change in https://github.com/apache/paimon/pull/4567/files.

Hi @wwj6591812 . Yes we will rewrite it. In the PR you mentioned, MonitorFunction is renamed to MonitorSource, as it is not a SourceFunction any more.

[JingsongLi]

+1