(flink-cdc) branch master updated: [hotfix][tests] Fix unstable `testInitialStartupModeWithOpTs` case (#3809)

[ruanhang1993]

This is an automated email from the ASF dual-hosted git repository.

ruanhang1993 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink-cdc.git


The following commit(s) were added to refs/heads/master by this push:
     new 9f44c523e [hotfix][tests] Fix unstable 
`testInitialStartupModeWithOpTs` case (#3809)
9f44c523e is described below

commit 9f44c523e4670c209e22b4ea8fb82511baa5fb5a
Author: yuxiqian <34335406+yuxiq...@users.noreply.github.com>
AuthorDate: Tue Dec 24 19:37:13 2024 +0800

    [hotfix][tests] Fix unstable `testInitialStartupModeWithOpTs` case (#3809)
    
    Signed-off-by: yuxiqian <34335406+yuxiq...@users.noreply.github.com>
---
 .../mysql/source/MySqlPipelineITCase.java          | 24 +++++++++++++++-------
 1 file changed, 17 insertions(+), 7 deletions(-)

diff --git 
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
 
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
index 276a2f12d..06cfa5628 100644
--- 
a/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
+++ 
b/flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-mysql/src/test/java/org/apache/flink/cdc/connectors/mysql/source/MySqlPipelineITCase.java
@@ -397,17 +397,27 @@ public class MySqlPipelineITCase extends 
MySqlSourceTestBase {
                                         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());