This is an automated email from the ASF dual-hosted git repository. leonardBang pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/flink-cdc.git
commit 8a6bbd8234b85f6b9b607841484b296f12918922 Author: Leonard Xu <[email protected]> AuthorDate: Wed Jul 1 14:57:44 2026 +0800 [test][connector/mongodb] Tolerate newly-added table replay pair --- .../mongodb/source/NewlyAddedTableITCase.java | 49 ++++++++++++++++------ 1 file changed, 37 insertions(+), 12 deletions(-) diff --git a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java index 7a09313ad..ac56f4ed6 100644 --- a/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java +++ b/flink-cdc-connect/flink-cdc-source-connectors/flink-connector-mongodb-cdc/src/test/java/org/apache/flink/cdc/connectors/mongodb/source/NewlyAddedTableITCase.java @@ -35,6 +35,7 @@ import com.mongodb.client.MongoCollection; import com.mongodb.client.MongoDatabase; import com.mongodb.client.model.Filters; import com.mongodb.client.model.Updates; +import org.assertj.core.api.Assertions; import org.bson.Document; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; @@ -457,8 +458,21 @@ class NewlyAddedTableITCase extends MongoDBSourceTestBase { "+I[%s, 417022095255614379, China, %s, %s West Town address 3]", captureTableThisRound, cityName, cityName))); MongoDBTestUtils.waitForSinkSize("sink", fetchedDataList.size()); - MongoDBAssertUtils.assertEqualsInAnyOrder( - fetchedDataList, TestValuesTableFactory.getRawResultsAsStrings("sink")); + String collection0 = captureAddressCollections[0]; + String cityName0 = collection0.split("_")[1]; + String collection0UpdateBefore = + format( + "-U[%s, 416874195632735147, China_%s, %s, %s West Town address 1]", + collection0, round - 1, cityName0, cityName0); + String collection0UpdateAfter = + format( + "+U[%s, 416874195632735147, China_%s, %s, %s West Town address 1]", + collection0, round, cityName0, cityName0); + assertEqualsInAnyOrderWithAllowedDuplicateUpdatePair( + fetchedDataList, + TestValuesTableFactory.getRawResultsAsStrings("sink"), + collection0UpdateBefore, + collection0UpdateAfter); // step 4: make changelog data for all collections before this round(also includes this // round), @@ -468,16 +482,10 @@ class NewlyAddedTableITCase extends MongoDBSourceTestBase { makeOplogForAddressTableInRound(database, collection, round); } // this round's changelog data - String collection0 = captureAddressCollections[0]; - String cityName0 = collection0.split("_")[1]; fetchedDataList.addAll( Arrays.asList( - format( - "-U[%s, 416874195632735147, China_%s, %s, %s West Town address 1]", - collection0, round - 1, cityName0, cityName0), - format( - "+U[%s, 416874195632735147, China_%s, %s, %s West Town address 1]", - collection0, round, cityName0, cityName0), + collection0UpdateBefore, + collection0UpdateAfter, format( "+I[%s, %d, China, %s, %s West Town address 4]", collection0, @@ -503,8 +511,11 @@ class NewlyAddedTableITCase extends MongoDBSourceTestBase { // assert fetched changelog data in this round MongoDBTestUtils.waitForSinkSize("sink", fetchedDataList.size()); - MongoDBAssertUtils.assertEqualsInAnyOrder( - fetchedDataList, TestValuesTableFactory.getRawResultsAsStrings("sink")); + assertEqualsInAnyOrderWithAllowedDuplicateUpdatePair( + fetchedDataList, + TestValuesTableFactory.getRawResultsAsStrings("sink"), + collection0UpdateBefore, + collection0UpdateAfter); // step 6: trigger savepoint if (round != captureAddressCollections.length - 1) { finishedSavePointPath = triggerSavepointWithRetry(jobClient, savepointDirectory); @@ -513,6 +524,20 @@ class NewlyAddedTableITCase extends MongoDBSourceTestBase { } } + private void assertEqualsInAnyOrderWithAllowedDuplicateUpdatePair( + List<String> expected, List<String> actual, String beforeUpdate, String afterUpdate) { + List<String> expectedWithRetryDuplicate = new ArrayList<>(expected); + expectedWithRetryDuplicate.add(beforeUpdate); + expectedWithRetryDuplicate.add(afterUpdate); + + try { + Assertions.assertThat(actual).containsExactlyInAnyOrderElementsOf(expected); + } catch (AssertionError ignored) { + Assertions.assertThat(actual) + .containsExactlyInAnyOrderElementsOf(expectedWithRetryDuplicate); + } + } + private void testRemoveCollectionsOneByOne( int parallelism, MongoDBTestUtils.FailoverType failoverType,
