This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/dev/pr-13018-2aa67dbd370d1591255da9a51d8ef70f01373010 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
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 + } +}
