[hotfix][tests] Fix unstable testInitialStartupModeWithOpTs case (a…

…pache#3809) Signed-off-by: yuxiqian <34335406+yuxiqian@users.noreply.github.com>
ChaomingZhangCN · Jan 3, 2025 · 3399e45 · 3399e45
1 parent 090a173
commit 3399e45
Showing 1 changed file with 17 additions and 7 deletions.
diff --git a/...mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java b/...mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -397,17 +397,27 @@ public void testInitialStartupModeWithOpTs() throws Exception {
                                         BinaryStringData.fromString("c-21")
                                     })));
         }
+
+        int snapshotRecordsCount = expectedSnapshot.size();
+        int binlogRecordsCount = expectedBinlog.size();
+
+        // Ditto, CreateTableEvent might be emitted in multiple partitions.
         List<Event> actual =
-                fetchResults(events, 1 + expectedSnapshot.size() + expectedBinlog.size());
-        assertThat(actual.get(0)).isEqualTo(createTableEvent);
-        assertThat(actual.subList(1, 10))
-                .containsExactlyInAnyOrder(expectedSnapshot.toArray(new Event[0]));
-        for (int i = 0; i < expectedBinlog.size(); i++) {
+                fetchResultsExcept(
+                        events, snapshotRecordsCount + binlogRecordsCount, createTableEvent);
+
+        List<Event> actualSnapshotEvents = actual.subList(0, snapshotRecordsCount);
+        List<Event> actualBinlogEvents = actual.subList(snapshotRecordsCount, actual.size());
+
+        assertThat(actualSnapshotEvents).containsExactlyInAnyOrderElementsOf(expectedSnapshot);
+        assertThat(actualBinlogEvents).hasSize(binlogRecordsCount);
+
+        for (int i = 0; i < binlogRecordsCount; i++) {
             if (expectedBinlog.get(i) instanceof SchemaChangeEvent) {
-                assertThat(expectedBinlog.get(i)).isEqualTo(actual.get(10 + i));
+                assertThat(actualBinlogEvents.get(i)).isEqualTo(expectedBinlog.get(i));
             } else {
                 DataChangeEvent expectedEvent = (DataChangeEvent) expectedBinlog.get(i);
-                DataChangeEvent actualEvent = (DataChangeEvent) actual.get(10 + i);
+                DataChangeEvent actualEvent = (DataChangeEvent) actualBinlogEvents.get(i);
                 assertThat(actualEvent.op()).isEqualTo(expectedEvent.op());
                 assertThat(actualEvent.before()).isEqualTo(expectedEvent.before());
                 assertThat(actualEvent.after()).isEqualTo(expectedEvent.after());