[Fix][Connector-V2] Fix issue of out-of-order fields in cdc

Author: corgy-w

seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mysql-e2e/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mysql/AbstractMysqlCDCITBase.java

@@ -300,7 +300,6 @@ public void testMysqlCdcCheckDataWithNoPrimaryKey(TestContainer container) {
     public void testMysqlCdcWithColumnIncludeList(TestContainer container) {
         // Clear related content to ensure that multiple operations are not affected
         clearTable(MYSQL_DATABASE, SOURCE_TABLE_1);
-        clearTable(MYSQL_DATABASE, "mysql_cdc_e2e_sink_table_column_include");
 
         CompletableFuture.supplyAsync(
                 () -> {
@@ -315,20 +314,26 @@ public void testMysqlCdcWithColumnIncludeList(TestContainer container) {
         await().atMost(60000, TimeUnit.MILLISECONDS)
                 .untilAsserted(
                         () -> {
-                            log.info(
-                                    query(
-                                                    getColumnIncludeQuerySQL(
-                                                            MYSQL_DATABASE,
-                                                            "mysql_cdc_e2e_sink_table_column_include"))
-                                            .toString());
-                            Assertions.assertIterableEquals(
-                                    query(
-                                            getColumnIncludeSourceQuerySQL(
-                                                    MYSQL_DATABASE, SOURCE_TABLE_1)),
-                                    query(
-                                            getColumnIncludeQuerySQL(
-                                                    MYSQL_DATABASE,
-                                                    "mysql_cdc_e2e_sink_table_column_include")));
+                            // Check if table exists first
+                            if (tableExists(
+                                    MYSQL_DATABASE, "mysql_cdc_e2e_sink_table_column_include")) {
+                                log.info(
+                                        query(
+                                                        getColumnIncludeQuerySQL(
+                                                                MYSQL_DATABASE,
+                                                                "mysql_cdc_e2e_sink_table_column_include"))
+                                                .toString());
+                                Assertions.assertIterableEquals(
+                                        query(
+                                                getColumnIncludeSourceQuerySQL(
+                                                        MYSQL_DATABASE, SOURCE_TABLE_1)),
+                                        query(
+                                                getColumnIncludeQuerySQL(
+                                                        MYSQL_DATABASE,
+                                                        "mysql_cdc_e2e_sink_table_column_include")));
+                            } else {
+                                Assertions.fail("Sink table not created yet");
+                            }
                         });
 
         // insert update delete
@@ -870,4 +875,13 @@ private void upsertDeleteSourceTableColumnInclude(String database, String tableN
         executeSql("DELETE FROM " + database + "." + tableName + " where id = 2");
         executeSql("UPDATE " + database + "." + tableName + " SET f_bigint = 10000 where id = 3");
     }
+
+    private boolean tableExists(String database, String tableName) {
+        try {
+            executeSql("SELECT 1 FROM " + database + "." + tableName + " LIMIT 1");
+            return true;
+        } catch (Exception e) {
+            return false;
+        }
+    }
 }