[FLINK-34688][cdc-connector][mysql] Make scan newly table trigger condition strictly

This closes #3519.

flink-cdc-connect/flink-cdc-source-connectors/flink-cdc-base/src/main/java/org/apache/flink/cdc/connectors/base/source/assigner/SnapshotSplitAssigner.java

@@ -203,7 +203,8 @@ public class SnapshotSplitAssigner<C extends SourceConfig> implements SplitAssig
 
     private void captureNewlyAddedTables() {
         if (sourceConfig.isScanNewlyAddedTableEnabled()
-                && !sourceConfig.getStartupOptions().isSnapshotOnly()) {
+                && !sourceConfig.getStartupOptions().isSnapshotOnly()
+                && AssignerStatus.isAssigningFinished(assignerStatus)) {
             try {
                 // check whether we got newly added tables
                 final List<TableId> currentCapturedTables =

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/main/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssigner.java

@@ -215,7 +215,8 @@ public class MySqlSnapshotSplitAssigner implements MySqlSplitAssigner {
     private void captureNewlyAddedTables() {
         // Don't scan newly added table in snapshot mode.
         if (sourceConfig.isScanNewlyAddedTableEnabled()
-                && !sourceConfig.getStartupOptions().isSnapshotOnly()) {
+                && !sourceConfig.getStartupOptions().isSnapshotOnly()
+                && AssignerStatus.isAssigningFinished(assignerStatus)) {
             // check whether we got newly added tables
             try (JdbcConnection jdbc = DebeziumUtils.openJdbcConnection(sourceConfig)) {
                 final List<TableId> currentCapturedTables =

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mysql-cdc/src/test/java/org/apache/flink/cdc/connectors/mysql/source/assigners/MySqlSnapshotSplitAssignerTest.java

@@ -484,7 +484,7 @@ public class MySqlSnapshotSplitAssignerTest extends MySqlSourceTestBase {
     }
 
     @Test
-    public void testScanNewlyAddedTableStartFromCheckpoint() {
+    public void testScanNewlyAddedTableStartFromInitialAssigningFinishedCheckpoint() {
         List<String> expected =
                 Arrays.asList(
                         "customers_sparse_dist [109] null",
@@ -492,7 +492,24 @@ public class MySqlSnapshotSplitAssignerTest extends MySqlSourceTestBase {
                         "customers_even_dist [10] [18]",
                         "customers_even_dist [18] null",
                         "customer_card_single_line null null");
-        assertEquals(expected, getTestAssignSnapshotSplitsFromCheckpoint());
+        assertEquals(
+                expected,
+                getTestAssignSnapshotSplitsFromCheckpoint(
+                        AssignerStatus.INITIAL_ASSIGNING_FINISHED));
+    }
+
+    @Test
+    public void testScanNewlyAddedTableStartFromNewlyAddedAssigningSnapshotFinishedCheckpoint() {
+        List<String> expected =
+                Arrays.asList(
+                        "customers_sparse_dist [109] null",
+                        "customers_even_dist null [10]",
+                        "customers_even_dist [10] [18]",
+                        "customers_even_dist [18] null");
+        assertEquals(
+                expected,
+                getTestAssignSnapshotSplitsFromCheckpoint(
+                        AssignerStatus.NEWLY_ADDED_ASSIGNING_SNAPSHOT_FINISHED));
     }
 
     private List<String> getTestAssignSnapshotSplits(
@@ -536,7 +553,7 @@ public class MySqlSnapshotSplitAssignerTest extends MySqlSourceTestBase {
         return getSplitsFromAssigner(assigner);
     }
 
-    private List<String> getTestAssignSnapshotSplitsFromCheckpoint() {
+    private List<String> getTestAssignSnapshotSplitsFromCheckpoint(AssignerStatus assignerStatus) {
         TableId newTable =
                 TableId.parse(customerDatabase.getDatabaseName() + ".customer_card_single_line");
         TableId processedTable =
@@ -619,7 +636,7 @@ public class MySqlSnapshotSplitAssignerTest extends MySqlSourceTestBase {
                         assignedSplits,
                         new HashMap<>(),
                         splitFinishedOffsets,
-                        AssignerStatus.INITIAL_ASSIGNING,
+                        assignerStatus,
                         remainingTables,
                         false,
                         true,