This is an automated email from the ASF dual-hosted git repository.
github-merge-queue[bot] pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/seatunnel.git
The following commit(s) were added to refs/heads/dev by this push:
new 69fbd2a84b [Fix][Connector-V2] Fix MongoDB CDC non-exactly-once
snapshot deadlock (#13018)
69fbd2a84b is described below
commit 69fbd2a84b7867b00e3dadb317495ae39f5b9366
Author: zhiweiniu <[email protected]>
AuthorDate: Wed Oct 28 05:06:18 2026 +0000
[Fix][Connector-V2] Fix MongoDB CDC non-exactly-once snapshot deadlock
(#13018)
---
docs/en/connectors/source/MongoDB-CDC.md | 2 +-
docs/zh/connectors/source/MongoDB-CDC.md | 2 +-
.../mongodb/source/fetch/MongodbScanFetchTask.java | 51 +++---
.../source/fetch/MongodbScanFetchTaskTest.java | 113 ++++++++++++++
.../src/test/java/mongodb/MongodbCDCIT.java | 172 +++++++++++++++++++++
...ongodbcdc_snapshot_with_concurrent_changes.conf | 63 ++++++++
6 files changed, 382 insertions(+), 21 deletions(-)
diff --git a/docs/en/connectors/source/MongoDB-CDC.md
b/docs/en/connectors/source/MongoDB-CDC.md
index 7ce4671312..a891e47049 100644
--- a/docs/en/connectors/source/MongoDB-CDC.md
+++ b/docs/en/connectors/source/MongoDB-CDC.md
@@ -132,7 +132,7 @@ For specific types in MongoDB, we use Extended JSON format
to map them to Seatun
| startup.timestamp | Long | No | - | Start
from the specified epoch timestamp in milliseconds. Only used when
`startup.mode` is `timestamp`.
|
| stop.mode | Enum | No | NEVER | Optional
stop mode for MongoDB CDC consumer. Valid values are `never` and `timestamp`.
See the [Stop Mode](#stop-mode) section below.
|
| stop.timestamp | Long | No | - | Stop at
the change-stream position derived from this epoch timestamp in milliseconds.
Only used when `stop.mode` is `timestamp`.
|
-| exactly_once | Boolean | No | false | Enable
exactly-once semantics. Enabling this may increase memory usage during large
table snapshot recovery.
|
+| exactly_once | Boolean | No | false | Enable
exactly-once semantics. When false, snapshot backfill is skipped and changes
made during the snapshot are read in the incremental phase; duplicate records
may be emitted. When true, snapshot backfill merges overlapping changes and may
increase memory usage. |
| format | Enum | No | DEFAULT | The
output data format of MongoDB CDC. Valid values are `DEFAULT` and
`COMPATIBLE_DEBEZIUM_JSON`.
|
| debezium | Config | No |
{key.converter.schemas.enable=false, value.converter.schemas.enable=false} |
Pass-through Debezium properties used by the embedded engine.
|
| common-options | | No | - | Source
plugin common parameters. For details, see [Source Common
Options](../common-options/source-common-options.md).
|
diff --git a/docs/zh/connectors/source/MongoDB-CDC.md
b/docs/zh/connectors/source/MongoDB-CDC.md
index 52f54dfd7d..60efe7340c 100644
--- a/docs/zh/connectors/source/MongoDB-CDC.md
+++ b/docs/zh/connectors/source/MongoDB-CDC.md
@@ -132,7 +132,7 @@ db.grantRolesToUser("<USER_NAME>", ["<ROLE_NAME>"])
| startup.timestamp | Long | 否 | - |
从指定的毫秒级时间戳开始消费。仅在 `startup.mode` 为 `timestamp` 时使用。
|
| stop.mode | Enum | 否 | NEVER | MongoDB CDC
的停止模式，可选值为 `never` 和 `timestamp`。详见下方[停止模式](#停止模式)。
|
| stop.timestamp | Long | 否 | - |
在该毫秒级时间戳对应的变更流位置停止。仅在 `stop.mode` 为 `timestamp` 时使用。
|
-| exactly_once | Boolean | 否 | false |
启用精确一次语义。开启后，大表快照阶段恢复时可能增加内存使用。
|
+| exactly_once | Boolean | 否 | false | 启用精确一次语义。设为
false 时跳过快照回填，快照期间的变更在增量阶段读取，可能输出重复记录。设为 true 时通过快照回填合并重叠变更，可能增加内存使用。 |
| format | Enum | 否 | DEFAULT | MongoDB CDC
的输出数据格式，可选值为 `DEFAULT` 和 `COMPATIBLE_DEBEZIUM_JSON`。
|
| debezium | Config | 否 |
{key.converter.schemas.enable=false, value.converter.schemas.enable=false} |
透传给内嵌 Debezium 引擎的配置。
|
| common-options | | 否 | - | 源插件通用参数，请参考
[源通用选项](../common-options/source-common-options.md)。
|
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTask.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTask.java
index 9bd9cabb70..6bff4e1132 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTask.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTask.java
@@ -129,27 +129,40 @@ public class MongodbScanFetchTask implements
FetchTask<SourceSplitBase> {
WatermarkKind.HIGH,
highWatermark)));
- log.info(
- "Snapshot step 4 - Back fill stream split for snapshot
split {}",
- snapshotSplit);
- final IncrementalSplit dataBackfillSplit =
- createBackfillStreamSplit(lowWatermark, highWatermark);
- final boolean streamBackfillRequired =
-
dataBackfillSplit.getStopOffset().isAfter(dataBackfillSplit.getStartupOffset());
+ if (taskContext.isExactlyOnce()) {
+ log.info(
+ "Snapshot step 4 - Back fill stream split for snapshot
split {}",
+ snapshotSplit);
+ final IncrementalSplit dataBackfillSplit =
+ createBackfillStreamSplit(lowWatermark, highWatermark);
+ final boolean streamBackfillRequired =
+ dataBackfillSplit
+ .getStopOffset()
+ .isAfter(dataBackfillSplit.getStartupOffset());
- if (!streamBackfillRequired) {
- changeEventQueue.enqueue(
- new DataChangeEvent(
- WatermarkEvent.create(
-
createWatermarkPartitionMap(collectionId.identifier()),
- "__mongodb_watermarks",
- dataBackfillSplit.splitId(),
- WatermarkKind.END,
- dataBackfillSplit.getStopOffset())));
+ if (!streamBackfillRequired) {
+ changeEventQueue.enqueue(
+ new DataChangeEvent(
+ WatermarkEvent.create(
+
createWatermarkPartitionMap(collectionId.identifier()),
+ "__mongodb_watermarks",
+ dataBackfillSplit.splitId(),
+ WatermarkKind.END,
+
dataBackfillSplit.getStopOffset())));
+ } else {
+ MongodbStreamFetchTask dataBackfillTask =
+ new MongodbStreamFetchTask(dataBackfillSplit);
+ dataBackfillTask.execute(taskContext);
+ }
} else {
- MongodbStreamFetchTask dataBackfillTask =
- new MongodbStreamFetchTask(dataBackfillSplit);
- dataBackfillTask.execute(taskContext);
+ // For non-exactly-once mode, the consumer stops reading after
HIGH watermark,
+ // so backfill events would fill up the queue and cause a
deadlock.
+ // Skip backfill entirely — the END watermark is not required
here
+ // because pollSplitRecordsIfNotExactlyOnce() only watches for
HIGH,
+ // and the stream split already starts at the low watermark.
+ log.info(
+ "Snapshot step 4 - Skipping backfill for
non-exactly-once mode, split {}",
+ snapshotSplit);
}
} catch (Exception e) {
throw new MongodbConnectorException(
diff --git
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTaskTest.java
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTaskTest.java
index 9da1f0b8dd..de223e0004 100644
---
a/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTaskTest.java
+++
b/seatunnel-connectors-v2/connector-cdc/connector-cdc-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/cdc/mongodb/source/fetch/MongodbScanFetchTaskTest.java
@@ -20,36 +20,149 @@ package
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.fetch;
import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
import org.apache.seatunnel.api.table.type.SeaTunnelRowType;
import org.apache.seatunnel.connectors.cdc.base.source.split.SnapshotSplit;
+import
org.apache.seatunnel.connectors.cdc.base.source.split.wartermark.WatermarkEvent;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.config.MongodbSourceConfig;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.exception.MongodbConnectorException;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.dialect.MongodbDialect;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.source.offset.ChangeStreamOffset;
import
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.MongodbUtils;
+import org.bson.BsonDocument;
+import org.bson.BsonInt32;
import org.bson.BsonTimestamp;
import org.bson.RawBsonDocument;
+import org.bson.codecs.BsonDocumentCodec;
import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.ValueSource;
+import org.mockito.MockedConstruction;
import org.mockito.MockedStatic;
import com.mongodb.MongoException;
+import com.mongodb.client.FindIterable;
import com.mongodb.client.MongoClient;
import com.mongodb.client.MongoCollection;
+import com.mongodb.client.MongoCursor;
import io.debezium.connector.base.ChangeEventQueue;
import io.debezium.pipeline.DataChangeEvent;
import io.debezium.relational.TableId;
+import io.debezium.util.LoggingContext;
+
+import java.time.Duration;
+import java.util.List;
import static org.apache.seatunnel.api.table.type.BasicType.INT_TYPE;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.ChunkUtils.maxUpperBoundOfId;
import static
org.apache.seatunnel.connectors.seatunnel.cdc.mongodb.utils.ChunkUtils.minLowerBoundOfId;
+import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.RETURNS_SELF;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.mockConstruction;
import static org.mockito.Mockito.mockStatic;
+import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
class MongodbScanFetchTaskTest {
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ @Timeout(10)
+ void shouldSkipBackfillInNonExactlyOnceMode(boolean advancingWatermark)
throws Exception {
+ assertSnapshotWatermarks(false, advancingWatermark);
+ }
+
+ @ParameterizedTest
+ @ValueSource(booleans = {false, true})
+ @Timeout(10)
+ void shouldPreserveExactlyOnceBackfill(boolean advancingWatermark) throws
Exception {
+ assertSnapshotWatermarks(true, advancingWatermark);
+ }
+
+ private void assertSnapshotWatermarks(boolean exactlyOnce, boolean
advancingWatermark)
+ throws Exception {
+ TableId collectionId = new TableId("inventory", null, "products");
+ SnapshotSplit split =
+ new SnapshotSplit(
+ "inventory.products:0",
+ collectionId,
+ new SeaTunnelRowType(
+ new String[] {"_id"}, new
SeaTunnelDataType<?>[] {INT_TYPE}),
+ minLowerBoundOfId(),
+ maxUpperBoundOfId());
+ MongodbFetchTaskContext context = mock(MongodbFetchTaskContext.class);
+ MongodbSourceConfig config = mock(MongodbSourceConfig.class);
+ MongodbDialect dialect = mock(MongodbDialect.class);
+ MongoClient client = mock(MongoClient.class);
+ @SuppressWarnings("unchecked")
+ MongoCollection<RawBsonDocument> collection =
mock(MongoCollection.class);
+ @SuppressWarnings("unchecked")
+ FindIterable<RawBsonDocument> iterable = mock(FindIterable.class,
RETURNS_SELF);
+ @SuppressWarnings("unchecked")
+ MongoCursor<RawBsonDocument> cursor = mock(MongoCursor.class);
+ ChangeEventQueue<DataChangeEvent> queue =
+ new ChangeEventQueue.Builder<DataChangeEvent>()
+ .pollInterval(Duration.ofMillis(10))
+ .maxBatchSize(exactlyOnce ? 4 : 3)
+ // A fourth event would block a non-exactly-once
snapshot task.
+ .maxQueueSize(exactlyOnce ? 4 : 3)
+ .loggingContextSupplier(
+ () -> LoggingContext.forConnector("mongodb",
"test", "snapshot"))
+ .build();
+ ChangeStreamOffset low = new ChangeStreamOffset(new
BsonTimestamp(1000, 1));
+ ChangeStreamOffset high =
+ advancingWatermark ? new ChangeStreamOffset(new
BsonTimestamp(2000, 1)) : low;
+ when(context.getSourceConfig()).thenReturn(config);
+ when(context.getDialect()).thenReturn(dialect);
+ when(context.getMongoClient()).thenReturn(client);
+ when(context.getQueue()).thenReturn(queue);
+ when(context.isExactlyOnce()).thenReturn(exactlyOnce);
+ when(config.getHosts()).thenReturn("localhost:27017");
+ when(dialect.displayCurrentOffset(config)).thenReturn(low, high);
+ when(collection.find()).thenReturn(iterable);
+ when(iterable.cursor()).thenReturn(cursor);
+ when(cursor.hasNext()).thenReturn(true, false);
+ when(cursor.next())
+ .thenReturn(
+ new RawBsonDocument(
+ new BsonDocument("_id", new BsonInt32(1)),
+ new BsonDocumentCodec()));
+
+ try (MockedStatic<MongodbUtils> utils = mockStatic(MongodbUtils.class);
+ MockedConstruction<MongodbStreamFetchTask> backfill =
+ mockConstruction(MongodbStreamFetchTask.class)) {
+ utils.when(() ->
MongodbUtils.buildConnectionNamespacePrefix("localhost:27017"))
+ .thenReturn("localhost:27017");
+ utils.when(
+ () ->
+ MongodbUtils.getMongoCollection(
+ client, collectionId,
RawBsonDocument.class))
+ .thenReturn(collection);
+ MongodbScanFetchTask task = new MongodbScanFetchTask(split);
+ task.execute(context);
+
+ List<DataChangeEvent> events = queue.poll();
+ boolean requiresBackfill = exactlyOnce && advancingWatermark;
+ assertEquals(requiresBackfill ? 1 : 0,
backfill.constructed().size());
+ if (requiresBackfill) {
+ verify(backfill.constructed().get(0)).execute(context);
+ }
+ assertEquals(exactlyOnce && !advancingWatermark ? 4 : 3,
events.size());
+
assertTrue(WatermarkEvent.isLowWatermarkEvent(events.get(0).getRecord()));
+
assertFalse(WatermarkEvent.isWatermarkEvent(events.get(1).getRecord()));
+
assertTrue(WatermarkEvent.isHighWatermarkEvent(events.get(2).getRecord()));
+ if (events.size() == 4) {
+
assertTrue(WatermarkEvent.isEndWatermarkEvent(events.get(3).getRecord()));
+ }
+ verify(cursor).close();
+ assertFalse(task.isRunning());
+ }
+ }
+
@Test
void shouldPreserveSnapshotReadFailureCause() throws Exception {
TableId collectionId = new TableId("inventory", null, "products");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
index d55819da3b..30571636e9 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/java/mongodb/MongodbCDCIT.java
@@ -119,6 +119,8 @@ public class MongodbCDCIT extends TestSuiteBase implements
TestResource {
private static final String DEFAULT_CHECKPOINT_PATH =
"/tmp/seatunnel/checkpoint_snapshot";
+ private static final String SNAPSHOT_READ_FAIL_POINT =
"waitInFindBeforeMakingBatch";
+
private static final MySqlContainer MYSQL_CONTAINER =
createMySqlContainer();
// mysql sink table query sql
@@ -159,6 +161,10 @@ public class MongodbCDCIT extends TestSuiteBase implements
TestResource {
log.info("The second stage:Starting Mongodb containers...");
mongodbContainer = new MongoDBContainer(NETWORK);
+ mongodbContainer.withCommand(
+ MongoDBContainer.ShardingClusterRole.startupCommand(
+ MongoDBContainer.ShardingClusterRole.NONE)
+ + " --setParameter enableTestCommands=1");
mongodbContainer.withLogConsumer(
new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("Mongodb-Docker-Image")));
@@ -168,6 +174,172 @@ public class MongodbCDCIT extends TestSuiteBase
implements TestResource {
log.info("Mongodb Container are started");
}
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "This regression test requires Zeta job status
and cancellation APIs")
+ public void
testMongodbCdcNonExactlyOnceSnapshotWithConcurrentChanges(TestContainer
container)
+ throws Exception {
+ assertSnapshotWithConcurrentChanges(container, false);
+ }
+
+ @TestTemplate
+ @DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason = "This regression test requires Zeta job status
and cancellation APIs")
+ public void
testMongodbCdcExactlyOnceSnapshotWithConcurrentChanges(TestContainer container)
+ throws Exception {
+ assertSnapshotWithConcurrentChanges(container, true);
+ }
+
+ private void assertSnapshotWithConcurrentChanges(TestContainer container,
boolean exactlyOnce)
+ throws Exception {
+ cleanSourceTable();
+ String collectionName = "snapshot_products_" + exactlyOnce;
+ MongoCollection<Document> products =
+
client.getDatabase(MONGODB_DATABASE).getCollection(collectionName);
+ products.drop();
+ int snapshotSize = 512;
+ List<Document> snapshotDocuments = new ArrayList<>();
+ for (int i = 0; i < snapshotSize; i++) {
+ snapshotDocuments.add(
+ new Document("_id", new ObjectId(String.format("%024x", i
+ 1)))
+ .append("name", "snapshot-" + i)
+ .append("description", "snapshot regression")
+ .append("weight", "1"));
+ }
+ products.insertMany(snapshotDocuments);
+ long failPointCount =
+ ((Number) configureSnapshotReadFailPoint(collectionName,
true).get("count"))
+ .longValue();
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return container.executeJob(
+
"/mongodbcdc_snapshot_with_concurrent_changes.conf",
+ jobId,
+ "collection=" + collectionName,
+ "exactly_once=" + exactlyOnce);
+ } catch (Exception e) {
+ throw new CompletionException(e);
+ }
+ });
+ try {
+ await().pollInterval(50, TimeUnit.MILLISECONDS)
+ .atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ () ->
+ "Snapshot job exited: "
+ +
jobFuture.join().getStderr());
+ Assertions.assertTrue(
+ snapshotWatermarkLogged(
+ container, collectionName,
"Snapshot step 1"));
+ });
+
+ client.getDatabase("admin")
+ .runCommand(
+ new Document("waitForFailPoint",
SNAPSHOT_READ_FAIL_POINT)
+ .append("timesEntered", failPointCount + 1)
+ .append("maxTimeMS", 10_000));
+ try {
+ // The snapshot find is paused after LOW. Generate more
changes than the
+ // 32-slot queue can hold before allowing the snapshot to
reach HIGH.
+ products.updateMany(
+ Filters.in(
+ "_id",
+ snapshotDocuments.subList(0, 128).stream()
+ .map(document ->
document.getObjectId("_id"))
+ .collect(Collectors.toList())),
+ Updates.set("weight", "2"));
+ Assertions.assertFalse(
+ snapshotWatermarkLogged(container, collectionName,
"Snapshot step 3"),
+ "Concurrent changes must finish before the snapshot
HIGH watermark");
+ } finally {
+ configureSnapshotReadFailPoint(collectionName, false);
+ }
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ () ->
+ "Snapshot job exited: "
+ +
jobFuture.join().getStderr());
+ Assertions.assertEquals(
+ Collections.singletonList(
+
Collections.singletonList((long) snapshotSize)),
+ querySql("select count(*) from
products"));
+ Assertions.assertEquals(
+
Collections.singletonList(Collections.singletonList(128L)),
+ querySql(
+ "select count(*) from products
where weight = '2'"));
+ });
+
+ products.insertOne(
+ new Document("_id", new
ObjectId("200000000000000000000001"))
+ .append("name", "after-snapshot")
+ .append("description", "incremental regression")
+ .append("weight", "1"));
+ products.updateOne(
+ Filters.eq("_id",
snapshotDocuments.get(0).getObjectId("_id")),
+ Updates.set("weight", "3"));
+ products.deleteOne(Filters.eq("_id",
snapshotDocuments.get(1).getObjectId("_id")));
+ await().atMost(60, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertEquals("RUNNING",
container.getJobStatus(jobId));
+ Assertions.assertEquals(
+ Collections.singletonList(
+
Collections.singletonList((long) snapshotSize)),
+ querySql("select count(*) from
products"));
+ Assertions.assertEquals(
+ Arrays.asList(
+
Arrays.asList("after-snapshot", "1"),
+ Arrays.asList("snapshot-0",
"3")),
+ querySql(
+ "select name, weight from
products "
+ + "where name in
('after-snapshot', 'snapshot-0', 'snapshot-1') order by name"));
+ });
+ } finally {
+ try {
+ configureSnapshotReadFailPoint(collectionName, false);
+ if (!jobFuture.isDone()) {
+ Container.ExecResult cancelled =
container.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelled.getExitCode(),
cancelled.getStderr());
+ }
+ jobFuture.get(60, TimeUnit.SECONDS);
+ } finally {
+ products.drop();
+ }
+ }
+ }
+
+ /** Pauses snapshot reads of the test collection until concurrent writes
have completed. */
+ private Document configureSnapshotReadFailPoint(String collectionName,
boolean enabled) {
+ return client.getDatabase("admin")
+ .runCommand(
+ new Document("configureFailPoint",
SNAPSHOT_READ_FAIL_POINT)
+ .append("mode", enabled ? "alwaysOn" : "off")
+ .append(
+ "data",
+ new Document("nss", MONGODB_DATABASE +
"." + collectionName)
+
.append("shouldCheckForInterrupt", true)));
+ }
+
+ private boolean snapshotWatermarkLogged(
+ TestContainer container, String collectionName, String step) {
+ return Arrays.stream(container.getServerLogs().split("\n"))
+ .anyMatch(line -> line.contains(step) &&
line.contains(collectionName));
+ }
+
@TestTemplate
public void testMongodbCdcToMysqlCheckDataE2e(TestContainer container)
throws InterruptedException {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/resources/mongodbcdc_snapshot_with_concurrent_changes.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/resources/mongodbcdc_snapshot_with_concurrent_changes.conf
new file mode 100644
index 0000000000..42f947751f
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-cdc-mongodb-e2e/src/test/resources/mongodbcdc_snapshot_with_concurrent_changes.conf
@@ -0,0 +1,63 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one or more
+# contributor license agreements. See the NOTICE file distributed with
+# this work for additional information regarding copyright ownership.
+# The ASF licenses this file to You under the Apache License, Version 2.0
+# (the "License"); you may not use this file except in compliance with
+# the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+env {
+ parallelism = 1
+ job.mode = "STREAMING"
+ checkpoint.interval = 5000
+}
+
+source {
+ MongoDB-CDC {
+ hosts = "mongo0:27017"
+ database = ["inventory"]
+ collection = ["inventory."${collection}]
+ exactly_once = ${exactly_once}
+ batch.size = 32
+ poll.max.batch.size = 16
+ poll.await.time.ms = 10
+ username = superuser
+ password = superpw
+ schema = {
+ primaryKey {
+ name = "id"
+ columnNames = ["_id"]
+ }
+ fields {
+ "_id": string,
+ "name": string,
+ "description": string,
+ "weight": string
+ }
+ }
+ }
+}
+
+sink {
+ jdbc {
+ url = "jdbc:mysql://mysql_e2e:3306/mongodb_cdc"
+ driver = "com.mysql.cj.jdbc.Driver"
+ username = "st_user"
+ password = "seatunnel"
+ generate_sink_sql = true
+ # You need to configure both database and table
+ database = mongodb_cdc
+ table = products
+ primary_keys = ["_id"]
+ batch_size = 1
+ }
+}
