This is an automated email from the ASF dual-hosted git repository.
davidzollo 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 15ebb211e1 [Feature][Connector-V2] Add MongoDB timer flush (#11527)
15ebb211e1 is described below
commit 15ebb211e1dd87660d4ea7073caa5057c0db3c0c
Author: zhiwei.niu <[email protected]>
AuthorDate: Sat Jul 25 20:44:36 2026 +0800
[Feature][Connector-V2] Add MongoDB timer flush (#11527)
---
docs/en/connectors/sink/MongoDB.md | 27 ++-
docs/zh/connectors/sink/MongoDB.md | 27 ++-
.../seatunnel/mongodb/sink/MongodbWriter.java | 12 +-
.../seatunnel/mongodb/sink/MongodbWriterTest.java | 142 ++++++++++++
.../connector-mongodb-e2e/pom.xml | 23 ++
.../e2e/connector/v2/mongodb/MongodbIT.java | 7 +-
.../connector/v2/mongodb/MongodbTimerFlushIT.java | 257 +++++++++++++++++++++
.../src/test/resources/ddl/shop.sql | 38 +++
.../src/test/resources/docker/server-gtids/my.cnf | 31 +++
.../src/test/resources/docker/setup.sql | 21 ++
.../resources/mysqlcdc_to_mongodb_timer_flush.conf | 49 ++++
11 files changed, 625 insertions(+), 9 deletions(-)
diff --git a/docs/en/connectors/sink/MongoDB.md
b/docs/en/connectors/sink/MongoDB.md
index dc603292f5..5e186c2055 100644
--- a/docs/en/connectors/sink/MongoDB.md
+++ b/docs/en/connectors/sink/MongoDB.md
@@ -13,6 +13,7 @@ import ChangeLog from '../changelog/connector-mongodb.md';
## Key features
- [x] [exactly-once](../../introduction/concepts/connector-v2-features.md)
+- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
(Zeta engine only)
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
@@ -81,10 +82,34 @@ The following table lists the field data type mapping from
MongoDB BSON type to
### Tips
-> 1.The data flushing logic of the MongoDB Sink Connector is jointly
controlled by three parameters: `buffer-flush.max-rows`,
`buffer-flush.interval`, and `checkpoint.interval`.<br/>
+> 1.The connector-level data flushing logic of MongoDB Sink is jointly
controlled by three parameters: `buffer-flush.max-rows`,
`buffer-flush.interval`, and `checkpoint.interval`.<br/>
> Data flushing will be triggered if any of these conditions are met.<br/>
> 2.Compatible with the historical parameter `upsert-key`. If `upsert-key` is
> set, please do not set `primary-key`.<br/>
+### Zeta Timer Flush
+
+This engine-level feature is supported only by Zeta. Spark and Flink do not
inject `FlushSignal` records. On Zeta, configure `sink.flush.interval` in the
`env` block to flush pending bulk requests even when `buffer-flush.max-rows`
has not been reached. Unlike `buffer-flush.interval`, the engine timer does not
require a new input record to trigger the check.
+
+Timer flush is enabled only when `transaction = false`. MongoDB transaction
mode is committed through checkpoints, so timer flush is disabled to preserve
the transaction boundary. The initial timer-flush implementation provides
at-least-once delivery rather than 2PC exactly-once. Enabling upsert with
deterministic primary keys can make retries idempotent.
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://127.0.0.1:27017"
+ database = "test_db"
+ collection = "users"
+ buffer-flush.max-rows = 10000
+ transaction = false
+ }
+}
+```
+
## How to Create a MongoDB Data Synchronization Jobs
The following example demonstrates how to create a data synchronization job
that writes randomly generated data to a MongoDB database:
diff --git a/docs/zh/connectors/sink/MongoDB.md
b/docs/zh/connectors/sink/MongoDB.md
index eb0411e913..e97ccaa10e 100644
--- a/docs/zh/connectors/sink/MongoDB.md
+++ b/docs/zh/connectors/sink/MongoDB.md
@@ -13,6 +13,7 @@ import ChangeLog from '../changelog/connector-mongodb.md';
## 关键特性
- [x] [exactly-once
精准一次写入](../../introduction/concepts/connector-v2-features.md)
+- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)（仅 Zeta 引擎）
- [x] [CDC（变更数据捕获）](../../introduction/concepts/connector-v2-features.md)
- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
@@ -80,10 +81,34 @@ MongoDB 连接器提供从 MongoDB 读取数据以及向 MongoDB 写入数据的
### 提示
-> 1. MongoDB Sink
连接器的数据刷新逻辑由以下三个参数共同控制：`buffer-flush.max-rows`、`buffer-flush.interval` 和
`checkpoint.interval`。
+> 1. MongoDB Sink
的连接器级数据刷新逻辑由以下三个参数共同控制：`buffer-flush.max-rows`、`buffer-flush.interval` 和
`checkpoint.interval`。<br/>
> 任一条件满足时，都会触发数据刷写。<br/>
> 2. 兼容历史参数 `upsert-key`。若已设置 `upsert-key`，请勿同时设置 `primary-key`。
+### Zeta 定时刷新
+
+该引擎级能力仅由 Zeta 支持，Spark 和 Flink 不会注入 `FlushSignal`。在 Zeta 中，可以在 `env` 块配置
`sink.flush.interval`，使未达到 `buffer-flush.max-rows` 的待处理 bulk 请求也能定时写出。与
`buffer-flush.interval` 不同，引擎定时器不需要新记录到达才触发检查。
+
+定时刷新仅在 `transaction = false` 时启用。MongoDB 事务模式通过 checkpoint
提交，因此会禁用定时刷新以保持事务边界。初始定时刷新实现提供至少一次语义，不提供基于 2PC 的精确一次语义。启用 upsert
并使用确定性主键可使重试具备幂等性。
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://127.0.0.1:27017"
+ database = "test_db"
+ collection = "users"
+ buffer-flush.max-rows = 10000
+ transaction = false
+ }
+}
+```
+
## 如何创建 MongoDB 数据同步任务
下面示例展示了一个将随机生成的数据写入 MongoDB 的数据同步任务：
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
index 14fe86a696..d0d356f10c 100644
---
a/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
+++
b/seatunnel-connectors-v2/connector-mongodb/src/main/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriter.java
@@ -70,17 +70,16 @@ public class MongodbWriter
private boolean transaction;
- // TODO：Reserve parameters.
- private final SinkWriter.Context context;
-
public MongodbWriter(
DocumentSerializer<SeaTunnelRow> serializer,
MongodbWriterOptions options,
SinkWriter.Context context) {
initOptions(options);
- this.context = context;
this.serializer = serializer;
this.bulkRequests = new ArrayList<>();
+ if (!transaction) {
+ context.registerFlushAction(this::timerFlush);
+ }
}
private void initOptions(MongodbWriterOptions options) {
@@ -131,6 +130,11 @@ public class MongodbWriter
return Optional.of(new MongodbCommitInfo(bsonDocuments));
}
+ /** Flushes pending bulk requests when the Zeta engine delivers a timer
flush signal. */
+ private void timerFlush() {
+ doBulkWrite();
+ }
+
private BsonDocument convertModelToBsonDocument(WriteModel<BsonDocument>
model) {
if (model instanceof InsertOneModel) {
return ((InsertOneModel<BsonDocument>) model).getDocument();
diff --git
a/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriterTest.java
b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriterTest.java
new file mode 100644
index 0000000000..a4d1cd83c2
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-mongodb/src/test/java/org/apache/seatunnel/connectors/seatunnel/mongodb/sink/MongodbWriterTest.java
@@ -0,0 +1,142 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.connectors.seatunnel.mongodb.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.type.RowKind;
+import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
+import
org.apache.seatunnel.connectors.seatunnel.mongodb.exception.MongodbConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.mongodb.internal.MongodbSingleCollectionProvider;
+import
org.apache.seatunnel.connectors.seatunnel.mongodb.serde.DocumentSerializer;
+
+import org.bson.BsonDocument;
+import org.bson.BsonInt32;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
+import org.mockito.MockedConstruction;
+import org.mockito.Mockito;
+
+import com.mongodb.MongoException;
+import com.mongodb.client.MongoCollection;
+import com.mongodb.client.model.BulkWriteOptions;
+import com.mongodb.client.model.InsertOneModel;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+class MongodbWriterTest {
+
+ @Test
+ void shouldRegisterAndExecuteTimerFlushForNonTransactionWriter() throws
Exception {
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ MongoCollection<BsonDocument> collection = mock(MongoCollection.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+
+ try (MockedConstruction<MongodbSingleCollectionProvider> ignored =
+ mockCollectionProvider(collection)) {
+ MongodbWriter writer = createWriter(context, false);
+ writer.write(createRow());
+
+ verify(context,
times(1)).registerFlushAction(actionCaptor.capture());
+ verify(collection, never())
+ .bulkWrite(Mockito.anyList(),
Mockito.any(BulkWriteOptions.class));
+
+ actionCaptor.getValue().run();
+
+ verify(collection, times(1))
+ .bulkWrite(Mockito.anyList(),
Mockito.any(BulkWriteOptions.class));
+ }
+ }
+
+ @Test
+ void shouldNotRegisterTimerFlushForTransactionWriter() {
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ MongoCollection<BsonDocument> collection = mock(MongoCollection.class);
+
+ try (MockedConstruction<MongodbSingleCollectionProvider> ignored =
+ mockCollectionProvider(collection)) {
+ createWriter(context, true);
+
+ verify(context, never()).registerFlushAction(Mockito.any());
+ }
+ }
+
+ @Test
+ void shouldPropagateTimerFlushFailure() throws Exception {
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ MongoCollection<BsonDocument> collection = mock(MongoCollection.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+ MongoException expected = new MongoException("timer flush failed");
+ doThrow(expected)
+ .when(collection)
+ .bulkWrite(Mockito.anyList(),
Mockito.any(BulkWriteOptions.class));
+
+ try (MockedConstruction<MongodbSingleCollectionProvider> ignored =
+ mockCollectionProvider(collection)) {
+ MongodbWriter writer = createWriter(context, false);
+ writer.write(createRow());
+ verify(context).registerFlushAction(actionCaptor.capture());
+
+ MongodbConnectorException actual =
+ Assertions.assertThrows(
+ MongodbConnectorException.class,
actionCaptor.getValue()::run);
+
+ Assertions.assertSame(expected, actual.getCause());
+ }
+ }
+
+ private MockedConstruction<MongodbSingleCollectionProvider>
mockCollectionProvider(
+ MongoCollection<BsonDocument> collection) {
+ return Mockito.mockConstruction(
+ MongodbSingleCollectionProvider.class,
+ (provider, context) ->
+
when(provider.getDefaultCollection()).thenReturn(collection));
+ }
+
+ private MongodbWriter createWriter(SinkWriter.Context context, boolean
transaction) {
+ DocumentSerializer<SeaTunnelRow> serializer =
mock(DocumentSerializer.class);
+ when(serializer.serializeToWriteModel(Mockito.any()))
+ .thenReturn(new InsertOneModel<>(new BsonDocument("id", new
BsonInt32(1))));
+ MongodbWriterOptions options =
+ MongodbWriterOptions.builder()
+ .withConnectString("mongodb://localhost:27017")
+ .withDatabase("timer_flush")
+ .withCollection("timer_flush")
+ .withFlushSize(100)
+ .withBatchIntervalMs(-1L)
+ .withRetryMax(0)
+ .withRetryInterval(0L)
+ .withTransaction(transaction)
+ .build();
+ return new MongodbWriter(serializer, options, context);
+ }
+
+ private SeaTunnelRow createRow() {
+ SeaTunnelRow row = mock(SeaTunnelRow.class);
+ when(row.getRowKind()).thenReturn(RowKind.INSERT);
+ return row;
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/pom.xml
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/pom.xml
index 35439ba63c..257f82858e 100644
--- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/pom.xml
+++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/pom.xml
@@ -25,6 +25,10 @@
<artifactId>connector-mongodb-e2e</artifactId>
<name>SeaTunnel : E2E : Connector V2 : Mongodb</name>
+ <properties>
+ <mysql.connector.version>8.0.32</mysql.connector.version>
+ </properties>
+
<dependencies>
<dependency>
<groupId>org.apache.seatunnel</groupId>
@@ -51,5 +55,24 @@
<type>test-jar</type>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>com.mysql</groupId>
+ <artifactId>mysql-connector-j</artifactId>
+ <version>${mysql.connector.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.seatunnel</groupId>
+ <artifactId>connector-cdc-mysql</artifactId>
+ <version>${project.version}</version>
+ <type>test-jar</type>
+ <scope>test</scope>
+ </dependency>
+ <dependency>
+ <groupId>org.testcontainers</groupId>
+ <artifactId>mysql</artifactId>
+ <version>${testcontainer.version}</version>
+ <scope>test</scope>
+ </dependency>
</dependencies>
</project>
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java
index 070b04c301..93f9663379 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbIT.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.e2e.connector.v2.mongodb;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.sink.DataSaveMode;
+import org.apache.seatunnel.api.sink.DefaultSinkWriterContext;
import org.apache.seatunnel.api.sink.SaveModeHandler;
import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
@@ -286,7 +287,7 @@ public class MongodbIT extends AbstractMongodbIT {
// build sink
final MongodbSink mongoDbSink = getSinkInstance(collectionName,
DataSaveMode.DROP_DATA);
final SinkWriter<SeaTunnelRow, MongodbCommitInfo, DocumentBulk> writer
=
- mongoDbSink.createWriter(null);
+ mongoDbSink.createWriter(new DefaultSinkWriterContext(0, 1));
final Optional<SaveModeHandler> saveModeHandlerOptional =
mongoDbSink.getSaveModeHandler();
// do save mode
if (saveModeHandlerOptional.isPresent()) {
@@ -315,7 +316,7 @@ public class MongodbIT extends AbstractMongodbIT {
// build sink
final MongodbSink mongoDbSink = getSinkInstance(collectionName,
DataSaveMode.APPEND_DATA);
final SinkWriter<SeaTunnelRow, MongodbCommitInfo, DocumentBulk> writer
=
- mongoDbSink.createWriter(null);
+ mongoDbSink.createWriter(new DefaultSinkWriterContext(0, 1));
final Optional<SaveModeHandler> saveModeHandlerOptional =
mongoDbSink.getSaveModeHandler();
// do save mode
if (saveModeHandlerOptional.isPresent()) {
@@ -345,7 +346,7 @@ public class MongodbIT extends AbstractMongodbIT {
final MongodbSink mongoDbSink =
getSinkInstance(collectionName,
DataSaveMode.ERROR_WHEN_DATA_EXISTS);
final SinkWriter<SeaTunnelRow, MongodbCommitInfo, DocumentBulk> writer
=
- mongoDbSink.createWriter(null);
+ mongoDbSink.createWriter(new DefaultSinkWriterContext(0, 1));
final Optional<SaveModeHandler> saveModeHandlerOptional =
mongoDbSink.getSaveModeHandler();
// do save mode
if (saveModeHandlerOptional.isPresent()) {
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbTimerFlushIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbTimerFlushIT.java
new file mode 100644
index 0000000000..53f0ce7ce5
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/java/org/apache/seatunnel/e2e/connector/v2/mongodb/MongodbTimerFlushIT.java
@@ -0,0 +1,257 @@
+/*
+ * 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.
+ */
+
+package org.apache.seatunnel.e2e.connector.v2.mongodb;
+
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlContainer;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.MySqlVersion;
+import
org.apache.seatunnel.connectors.seatunnel.cdc.mysql.testutils.UniqueDatabase;
+import org.apache.seatunnel.e2e.common.TestResource;
+import org.apache.seatunnel.e2e.common.TestSuiteBase;
+import org.apache.seatunnel.e2e.common.container.ContainerExtendedFactory;
+import org.apache.seatunnel.e2e.common.container.EngineType;
+import org.apache.seatunnel.e2e.common.container.TestContainer;
+import org.apache.seatunnel.e2e.common.junit.DisabledOnContainer;
+import org.apache.seatunnel.e2e.common.junit.TestContainerExtension;
+import org.apache.seatunnel.e2e.common.util.JobIdGenerator;
+
+import org.bson.Document;
+import org.junit.jupiter.api.AfterAll;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeAll;
+import org.junit.jupiter.api.TestTemplate;
+import org.testcontainers.containers.Container;
+import org.testcontainers.containers.GenericContainer;
+import org.testcontainers.containers.output.Slf4jLogConsumer;
+import org.testcontainers.containers.wait.strategy.Wait;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerImageName;
+import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
+
+import com.mongodb.client.MongoClient;
+import com.mongodb.client.MongoClients;
+import lombok.extern.slf4j.Slf4j;
+
+import java.io.IOException;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.sql.Connection;
+import java.sql.Statement;
+import java.time.Duration;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "engine-level timer flush (sink.flush.interval) is only
supported on Zeta engine")
+public class MongodbTimerFlushIT extends TestSuiteBase implements TestResource
{
+
+ private static final String MONGODB_IMAGE = "mongo:latest";
+ private static final String MONGODB_HOST = "mongodb_timer_flush_e2e";
+ private static final int MONGODB_PORT = 27017;
+ private static final String MONGODB_DATABASE = "test_db";
+ private static final String MONGODB_COLLECTION = "mongodb_timer_flush";
+ private static final String MYSQL_HOST = "mysql_mongodb_timer_flush_e2e";
+ private static final String MYSQL_USER_NAME = "mysqluser";
+ private static final String MYSQL_USER_PASSWORD = "mysqlpw";
+ private static final String MYSQL_DATABASE = "shop";
+ private static final String MYSQL_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/MySQL-CDC/lib";
+ private static final MySqlContainer MYSQL_CONTAINER =
createMySqlContainer(MySqlVersion.V8_0);
+
+ private final UniqueDatabase shopDatabase = new
UniqueDatabase(MYSQL_CONTAINER, MYSQL_DATABASE);
+ private GenericContainer<?> mongodbContainer;
+ private MongoClient mongoClient;
+
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
this::copyMySQLDriverToContainer;
+
+ @BeforeAll
+ @Override
+ public void startUp() {
+ mongodbContainer =
+ new GenericContainer<>(DockerImageName.parse(MONGODB_IMAGE))
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MONGODB_HOST)
+ .withExposedPorts(MONGODB_PORT)
+ .waitingFor(
+
Wait.forListeningPort().withStartupTimeout(Duration.ofMinutes(2)))
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger(MONGODB_IMAGE)));
+ Startables.deepStart(Stream.of(mongodbContainer,
MYSQL_CONTAINER)).join();
+
+ await().ignoreExceptions()
+ .atMost(180, TimeUnit.SECONDS)
+ .untilAsserted(this::initializeMongoClient);
+ shopDatabase.createAndInitialize();
+
mongoClient.getDatabase(MONGODB_DATABASE).getCollection(MONGODB_COLLECTION).drop();
+ }
+
+ @TestTemplate
+ public void testMongodbTimerFlush(TestContainer testContainer) throws
Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return testContainer.executeJob(
+
"/mysqlcdc_to_mongodb_timer_flush.conf", jobId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ await().atMost(2, TimeUnit.MINUTES)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobStillRunning(
+ jobFuture,
+ "The streaming job terminated before
reaching RUNNING");
+ Assertions.assertEquals(
+ "RUNNING",
testContainer.getJobStatus(jobId));
+ });
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobStillRunning(
+ jobFuture,
+ "The streaming job terminated before
timer flush published the snapshot");
+ Assertions.assertEquals(9, documentCount());
+ });
+
+ try (Connection connection = shopDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.executeUpdate(
+ "INSERT INTO products (id, name, description, weight) "
+ + "VALUES (110, 'timer-flush', 'timer-flush
probe', 1.0)");
+ }
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ assertJobStillRunning(
+ jobFuture,
+ "The streaming job terminated before
timer flush published the binlog event");
+ Assertions.assertEquals(10, documentCount());
+ });
+ } finally {
+ if (!jobFuture.isDone()) {
+ Container.ExecResult cancelResult =
testContainer.cancelJob(jobId);
+ Assertions.assertEquals(0, cancelResult.getExitCode(),
cancelResult.getStderr());
+ }
+ }
+
+ Container.ExecResult jobResult = jobFuture.get(120, TimeUnit.SECONDS);
+ Assertions.assertEquals(0, jobResult.getExitCode(),
jobResult.getStderr());
+ }
+
+ private long documentCount() {
+ return mongoClient
+ .getDatabase(MONGODB_DATABASE)
+ .getCollection(MONGODB_COLLECTION)
+ .countDocuments();
+ }
+
+ private void assertJobStillRunning(
+ CompletableFuture<Container.ExecResult> jobFuture, String message)
throws Exception {
+ if (jobFuture.isDone()) {
+ Container.ExecResult jobResult = jobFuture.get();
+ Assertions.fail(message + ":\n" + jobResult.getStderr());
+ }
+ }
+
+ private void initializeMongoClient() {
+ if (mongoClient != null) {
+ mongoClient.close();
+ }
+ mongoClient =
+ MongoClients.create(
+ String.format(
+ "mongodb://%s:%d",
+ mongodbContainer.getHost(),
+ mongodbContainer.getMappedPort(MONGODB_PORT)));
+ mongoClient.getDatabase("admin").runCommand(new Document("ping", 1));
+ }
+
+ private void copyMySQLDriverToContainer(GenericContainer<?> container)
+ throws IOException, InterruptedException {
+ Path driverJarPath = mysqlDriverJarPath();
+ Assertions.assertTrue(
+ Files.isRegularFile(driverJarPath),
+ "MySQL JDBC driver should be resolved from the test classpath
before E2E runs: "
+ + driverJarPath);
+ Container.ExecResult extraCommands =
+ container.execInContainer("bash", "-c", "mkdir -p " +
MYSQL_CDC_PLUGIN_LIB);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ MYSQL_CDC_PLUGIN_LIB + "/" + driverJarPath.getFileName());
+ }
+
+ private Path mysqlDriverJarPath() {
+ try {
+ return Paths.get(
+ com.mysql.cj.jdbc.Driver.class
+ .getProtectionDomain()
+ .getCodeSource()
+ .getLocation()
+ .toURI());
+ } catch (Exception e) {
+ throw new RuntimeException(
+ "Failed to resolve MySQL JDBC driver jar from the test
classpath", e);
+ }
+ }
+
+ private static MySqlContainer createMySqlContainer(MySqlVersion version) {
+ return new MySqlContainer(version)
+ .withConfigurationOverride("docker/server-gtids/my.cnf")
+ .withSetupSQL("docker/setup.sql")
+ .withNetwork(NETWORK)
+ .withNetworkAliases(MYSQL_HOST)
+ .withDatabaseName(MYSQL_DATABASE)
+ .withUsername(MYSQL_USER_NAME)
+ .withPassword(MYSQL_USER_PASSWORD)
+ .withLogConsumer(
+ new
Slf4jLogConsumer(DockerLoggerFactory.getLogger("mysql-docker-image")));
+ }
+
+ @AfterAll
+ @Override
+ public void tearDown() {
+ if (mongoClient != null) {
+ mongoClient.close();
+ }
+ if (mongodbContainer != null) {
+ mongodbContainer.close();
+ }
+ MYSQL_CONTAINER.close();
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/ddl/shop.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/ddl/shop.sql
new file mode 100644
index 0000000000..253630e7ee
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/ddl/shop.sql
@@ -0,0 +1,38 @@
+--
+-- 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.
+--
+
+CREATE DATABASE IF NOT EXISTS `shop`;
+USE shop;
+
+DROP TABLE IF EXISTS products;
+CREATE TABLE products (
+ id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
+ name VARCHAR(255) NOT NULL DEFAULT 'SeaTunnel',
+ description VARCHAR(512),
+ weight FLOAT
+);
+
+INSERT INTO products
+VALUES (101,"scooter","Small 2-wheel scooter",3.14),
+ (102,"car battery","12V car battery",8.1),
+ (103,"12-pack drill bits","12-pack of drill bits with sizes ranging
from #40 to #3",0.8),
+ (104,"hammer","12oz carpenter's hammer",0.75),
+ (105,"hammer","14oz carpenter's hammer",0.875),
+ (106,"hammer","16oz carpenter's hammer",1.0),
+ (107,"rocks","box of assorted rocks",5.3),
+ (108,"jacket","water resistent black wind breaker",0.1),
+ (109,"spare tire","24 inch spare tire",22.2);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/server-gtids/my.cnf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/server-gtids/my.cnf
new file mode 100644
index 0000000000..bfe5641b9c
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/server-gtids/my.cnf
@@ -0,0 +1,31 @@
+#
+# 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.
+#
+
+[mysqld]
+skip-host-cache
+skip-name-resolve
+secure-file-priv=/var/lib/mysql
+user=mysql
+symbolic-links=0
+
+server-id = 223344
+log_bin = mysql-bin
+expire_logs_days = 1
+binlog_format = row
+
+gtid_mode = on
+enforce_gtid_consistency = on
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/setup.sql
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/setup.sql
new file mode 100644
index 0000000000..95b8e560b9
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/docker/setup.sql
@@ -0,0 +1,21 @@
+--
+-- 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.
+--
+
+GRANT ALL PRIVILEGES ON *.* TO 'mysqluser'@'%';
+
+CREATE USER 'st_user_source' IDENTIFIED BY 'mysqlpw';
+GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT,
DROP, LOCK TABLES ON *.* TO 'st_user_source'@'%';
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mysqlcdc_to_mongodb_timer_flush.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mysqlcdc_to_mongodb_timer_flush.conf
new file mode 100644
index 0000000000..eced0b66c2
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-mongodb-e2e/src/test/resources/mysqlcdc_to_mongodb_timer_flush.conf
@@ -0,0 +1,49 @@
+#
+# 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 = 300000
+ sink.flush.interval = 500
+}
+
+source {
+ MySQL-CDC {
+ parallelism = 1
+ server-id = 5692
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.products"]
+ url = "jdbc:mysql://mysql_mongodb_timer_flush_e2e:3306/shop"
+ table-names-config = [{"table": "shop.products", "primaryKeys": ["id"]}]
+ }
+}
+
+sink {
+ MongoDB {
+ uri = "mongodb://mongodb_timer_flush_e2e:27017/test_db"
+ database = "test_db"
+ collection = "mongodb_timer_flush"
+ buffer-flush.max-rows = 100000
+ buffer-flush.interval = -1
+ upsert-enable = true
+ primary-key = ["id"]
+ transaction = false
+ data_save_mode = "APPEND_DATA"
+ }
+}
