[Mysql] Fix duplicate split which cause duplicate data when open scanNewlyAddedTableEnabled

[EMsnap]

Fix #2095

I carefully debugged the bug and there is the analysis:

1、In MySqlSnapshotSplitAssigner, split table happens asynchronously
2、After the finish of every table split the table will be moved from remaining table
3、After tm calls getNext() from MySqlSnapshotSplitAssigner, the table will be added to alreadyProcessedTables
4、However when job restarts and open scanNewlyAddedTableEnabled and discover newly added table, it gets into the following logic

The alreadyProcessedTables + remainingTables won't always be euqals to the tables splitted in the last run.

This happens when table is so large and tm can't process split on time.

For example:
1、The task reads from A and B, and each of them split into 1000 chunks
2、The remaingTable is empty since they are already splitted, but alreadyProcessedTables only contains A since B is not fetch by tm now
3、 job restarts and B will be spliited again.

[EMsnap]

@leonardBang @lzshlzsh PLAL

[ruanhang1993]

@EMsnap Thanks.
Please rebase the master branch.

[EMsnap]

@EMsnap Thanks. Please rebase the master branch.

Sure, done

Thanks for the update

[ruanhang1993]

@EMsnap Thanks. Please rebase the master branch.

Sure, done

Thanks for the update

Thanks @EMsnap . The CI failed. Please take a look at it.

[EMsnap]

@EMsnap Thanks. Please rebase the master branch.

Sure, done
Thanks for the update

Thanks @EMsnap . The CI failed. Please take a look at it.

I guess the failed UT is not related to the pr since the pr works on mysql-cdc, could u help rerun the UT ?

[ruanhang1993]

@EMsnap I think this PR should be modified. Actually all captured tables are not equals to remainingTables + alreadyProcessedTables. The method computeTablesPendingSnapshot may be helpful to understand it.

[ruanhang1993]

We should change the code in captureNewlyAddedTables instead of changing the location of addAlreadyProcessedTablesIfNotExists.

[EMsnap]

Noted, I'll change the implementation asap

[EMsnap]

@EMsnap I think this PR should be modified. Actually all captured tables are not equals to remainingTables + alreadyProcessedTables. The method computeTablesPendingSnapshot may be helpful to understand it.

Thanks for the reply, I'll take a look.
But where is this computeTablesPendingSnapshot method located? I can't find it on the master branch

[ruanhang1993]

computeTablesPendingSnapshot

Sorry, the computeTablesPendingSnapshot are not provided before the Flink 1.18. Please ignore that.

[ruanhang1993]

@EMsnap Thanks for the work. We try to fix this in the version 2.4.1. So I provide a new changes to fix this.

[ruanhang1993]

LGTM

[EMsnap]

@EMsnap Thanks for the work. We try to fix this in the version 2.4.1. So I provide a new changes to fix this.

Thanks for your reply and your new changes, great job !