[Bug] Flink bounded source with checkpoint missing last snapshot commit

[eric666666]

trafficstars

Search before asking

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

Paimon version

Paimon 0.9 snapshot

Compute Engine

Flink

Minimal reproduce step

If use flink bounded source which like jdbc or other, with checkpoint enabled,flink do not commit last snapshot commit. I review the source code find org.apache.paimon.flink.sink.CommitterOperator caused it. I do not know why their should skip commit snapshot if flink streaming mode enable checkpoint.

What doesn't meet your expectations?

Missing last commit

Anything else?

No response

Are you willing to submit a PR?

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

[eric666666]

@tsreaper @JingsongLi Please have a look.

[eric666666]

@tsreaper @JingsongLi Please have a look.

I remove the condition, to fix the problem. But I look at the code, the conditions here seem to be intended, so I don't know if I should do this

[JingsongLi]

Flink 1.14?

[eric666666]

Flink 1.14?

No, Flink 1.18.1

[LinMingQiang]

As I know, flink 15+ version has support checkpointing with finished tasks (https://flink.apache.org/2022/07/11/flip-147-support-checkpoints-after-tasks-finished-part-one/#support-checkpointing-with-finished-tasks). so commit will be executed once in notifyCheckpointComplete. And this was true in my tests.