[FLINK-36517][cdc-connect][paimon] use filterAndCommit API for Avoid commit the same datafile duplicate #3639
Conversation
|
@lvyanquan @leonardBang PTAL |
| "Commit succeeded for %s with %s committable", | ||
| checkpointId, committables.size())); | ||
| } catch (Exception e) { | ||
| commitRequests.forEach(CommitRequest::retryLater); |
There was a problem hiding this comment.
Is there a specific purpose for retrying later in this context? @lvyanquan
|
Could you please assist in reviewing this PR? Thank you. @lvyanquan |
|
I agree that the issue of duplicate commits still exists. Our testing in the case of abnormal failover is relatively lacking, can you try adding corresponding test case for this? |
I will try, thanks. |
…commit the same datafile duplicate
| // It's possible that flink job will restore from a checkpoint with only step#1 finished and | ||
| // step#2 not. | ||
| // CommitterOperator will try to re-commit recovered transactions. | ||
| committer.commit(commitRequests); |
There was a problem hiding this comment.
Thanks for adding this, what about running insert and commit many times(in a for loop), to simulate more complex situations and situations with compaction?
There was a problem hiding this comment.
Considering there is another issue https://issues.apache.org/jira/browse/FLINK-36541 in PaimonWriter, If there is a problem with adding this loop, you can skip it for now.
https://issues.apache.org/jira/browse/FLINK-35938 problem still persists.
storeMultiCommitter.commitAPI may cause the same datafile commit twice when job restart from failure.