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 25137c1d8f [Feature][Connector-V2] Add StarRocks timer flush (#11518)
25137c1d8f is described below
commit 25137c1d8fb8b2410eb0641343cac4e524f8375f
Author: zhiwei.niu <[email protected]>
AuthorDate: Sat Jul 25 21:35:49 2026 +0800
[Feature][Connector-V2] Add StarRocks timer flush (#11518)
---
docs/en/connectors/sink/StarRocks.md | 33 ++-
docs/zh/connectors/sink/StarRocks.md | 33 ++-
.../seatunnel/starrocks/sink/StarRocksSink.java | 2 +-
.../starrocks/sink/StarRocksSinkWriter.java | 11 +-
.../starrocks/sink/StarRocksSinkWriterTest.java | 92 +++++++
.../connector/starrocks/StarRocksTimerFlushIT.java | 276 +++++++++++++++++++++
.../mysqlcdc_to_starrocks_timer_flush.conf | 53 ++++
7 files changed, 496 insertions(+), 4 deletions(-)
diff --git a/docs/en/connectors/sink/StarRocks.md
b/docs/en/connectors/sink/StarRocks.md
index cd92b81d1b..2d22517a64 100644
--- a/docs/en/connectors/sink/StarRocks.md
+++ b/docs/en/connectors/sink/StarRocks.md
@@ -15,7 +15,7 @@ import ChangeLog from '../changelog/connector-starrocks.md';
- [ ] [exactly-once](../../introduction/concepts/connector-v2-features.md)
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
- [x] [support multiple table
write](../../introduction/concepts/connector-v2-features.md)
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
## Description
@@ -170,6 +170,37 @@ sink {
}
```
+### Zeta Timer Flush
+
+This engine-level capability is available only in Zeta. Configure
`sink.flush.interval` in `env` to periodically write buffered rows through
StarRocks Stream Load even when `batch_max_rows` and `batch_max_bytes` have not
been reached. Spark and Flink do not trigger this scheduled flush.
+
+:::tip
+
+StarRocks timer flush does not provide 2PC exactly-once semantics. The
StarRocks Sink remains at-least-once, and a task restart may submit rows again.
When appropriate for the workload, a Primary Key table with deterministic keys
can absorb duplicate writes.
+
+:::
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ StarRocks {
+ nodeUrls = ["starrocks-fe:8030"]
+ base-url = "jdbc:mysql://starrocks-fe:9030/mydb"
+ username = root
+ password = ""
+ database = "mydb"
+ table = "mytable"
+ batch_max_rows = 10000
+ batch_max_bytes = 104857600
+ }
+}
+```
+
## Data Type Mapping
| StarRocks Data type | SeaTunnel Data type |
diff --git a/docs/zh/connectors/sink/StarRocks.md
b/docs/zh/connectors/sink/StarRocks.md
index e3a27b9af9..32726cfff7 100644
--- a/docs/zh/connectors/sink/StarRocks.md
+++ b/docs/zh/connectors/sink/StarRocks.md
@@ -14,7 +14,7 @@ import ChangeLog from '../changelog/connector-starrocks.md';
- [ ] [精准一次](../../introduction/concepts/connector-v2-features.md)
- [x] [cdc](../../introduction/concepts/connector-v2-features.md)
-- [ ] [timer flush](../../introduction/concepts/connector-v2-features.md)
+- [x] [timer flush](../../introduction/concepts/connector-v2-features.md)
## 描述
@@ -164,6 +164,37 @@ sink {
}
```
+### Zeta 定时刷新
+
+该引擎级能力仅由 Zeta 支持。可以在 `env` 块中配置 `sink.flush.interval`，使尚未达到 `batch_max_rows` 和
`batch_max_bytes` 的缓冲数据也能定时通过 StarRocks Stream Load 写出。Spark 和 Flink 不会触发该定时刷新。
+
+:::tip
+
+StarRocks 定时刷新不提供基于 2PC 的精准一次语义，StarRocks Sink
仍为至少一次语义，任务重启后可能重复提交数据。如果业务场景适用，可以使用具有确定性主键的 Primary Key 表吸收重复写入。
+
+:::
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ StarRocks {
+ nodeUrls = ["starrocks-fe:8030"]
+ base-url = "jdbc:mysql://starrocks-fe:9030/mydb"
+ username = root
+ password = ""
+ database = "mydb"
+ table = "mytable"
+ batch_max_rows = 10000
+ batch_max_bytes = 104857600
+ }
+}
+```
+
## 数据类型映射
| StarRocks数据类型 | SeaTunnel数据类型 |
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
index b856ca1e64..7abdf06e3e 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSink.java
@@ -85,7 +85,7 @@ public class StarRocksSink extends
AbstractSimpleSink<SeaTunnelRow, Void>
log.warn("Failed to load JDBC driver {}",
"com.mysql.cj.jdbc.Driver", e);
}
TablePath sinkTablePath = catalogTable.getTablePath();
- return new StarRocksSinkWriter(sinkConfig, tableSchema, sinkTablePath);
+ return new StarRocksSinkWriter(context, sinkConfig, tableSchema,
sinkTablePath);
}
@Override
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java
index 902e288e3d..41df65a3fc 100644
---
a/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriter.java
@@ -17,6 +17,7 @@
package org.apache.seatunnel.connectors.seatunnel.starrocks.sink;
+import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.sink.SupportMultiTableSinkWriter;
import org.apache.seatunnel.api.sink.SupportSchemaEvolutionSinkWriter;
import org.apache.seatunnel.api.table.catalog.TablePath;
@@ -57,13 +58,17 @@ public class StarRocksSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
new TableSchemaChangeEventDispatcher();
public StarRocksSinkWriter(
- SinkConfig sinkConfig, TableSchema tableSchema, TablePath
tablePath) {
+ SinkWriter.Context context,
+ SinkConfig sinkConfig,
+ TableSchema tableSchema,
+ TablePath tablePath) {
this.tableSchema = tableSchema;
SeaTunnelRowType seaTunnelRowType =
tableSchema.toPhysicalRowDataType();
this.serializer = createSerializer(sinkConfig, seaTunnelRowType);
this.manager = new StarRocksSinkManager(sinkConfig, tableSchema);
this.sinkConfig = sinkConfig;
this.sinkTablePath = tablePath;
+ context.registerFlushAction(this::timerFlush);
}
@Override
@@ -110,6 +115,10 @@ public class StarRocksSinkWriter extends
AbstractSinkWriter<SeaTunnelRow, Void>
return super.prepareCommit();
}
+ private void timerFlush() throws IOException {
+ manager.flush();
+ }
+
@Override
public void close() throws IOException {
try {
diff --git
a/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriterTest.java
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriterTest.java
new file mode 100644
index 0000000000..e7fa042c0a
--- /dev/null
+++
b/seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/sink/StarRocksSinkWriterTest.java
@@ -0,0 +1,92 @@
+/*
+ * 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.starrocks.sink;
+
+import org.apache.seatunnel.api.sink.SinkWriter;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TablePath;
+import org.apache.seatunnel.api.table.catalog.TableSchema;
+import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
+import
org.apache.seatunnel.connectors.seatunnel.starrocks.client.StarRocksSinkManager;
+import org.apache.seatunnel.connectors.seatunnel.starrocks.config.SinkConfig;
+
+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 java.io.IOException;
+
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+
+public class StarRocksSinkWriterTest {
+
+ @Test
+ void shouldRegisterTimerFlushAction() throws Exception {
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+
+ try (MockedConstruction<StarRocksSinkManager> mockedManager =
+ Mockito.mockConstruction(StarRocksSinkManager.class)) {
+ createWriter(context);
+
+ verify(context,
times(1)).registerFlushAction(actionCaptor.capture());
+ actionCaptor.getValue().run();
+ verify(mockedManager.constructed().get(0), times(1)).flush();
+ }
+ }
+
+ @Test
+ void shouldPropagateTimerFlushFailure() throws Exception {
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+ ArgumentCaptor<RunnableWithException> actionCaptor =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+
+ try (MockedConstruction<StarRocksSinkManager> mockedManager =
+ Mockito.mockConstruction(StarRocksSinkManager.class)) {
+ createWriter(context);
+ StarRocksSinkManager manager = mockedManager.constructed().get(0);
+ IOException expected = new IOException("timer flush failed");
+ doThrow(expected).when(manager).flush();
+
+ verify(context).registerFlushAction(actionCaptor.capture());
+ IOException actual =
+ Assertions.assertThrows(IOException.class,
actionCaptor.getValue()::run);
+ Assertions.assertSame(expected, actual);
+ }
+ }
+
+ private StarRocksSinkWriter createWriter(SinkWriter.Context context) {
+ SinkConfig sinkConfig = new SinkConfig();
+ sinkConfig.setLoadFormat(SinkConfig.StreamLoadFormat.JSON);
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.INT_TYPE, (Long) null,
false, null, null))
+ .build();
+ return new StarRocksSinkWriter(
+ context, sinkConfig, tableSchema, TablePath.of("test",
"timer_flush"));
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksTimerFlushIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksTimerFlushIT.java
new file mode 100644
index 0000000000..5f472bd014
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/java/org/apache/seatunnel/e2e/connector/starrocks/StarRocksTimerFlushIT.java
@@ -0,0 +1,276 @@
+/*
+ * 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.starrocks;
+
+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.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.lifecycle.Startables;
+import org.testcontainers.utility.DockerLoggerFactory;
+import org.testcontainers.utility.MountableFile;
+
+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.DriverManager;
+import java.sql.ResultSet;
+import java.sql.SQLException;
+import java.sql.Statement;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.TimeUnit;
+import java.util.stream.Stream;
+
+import static org.awaitility.Awaitility.await;
+import static org.awaitility.Awaitility.given;
+
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "engine-level timer flush (sink.flush.interval) is only
supported on Zeta engine")
+public class StarRocksTimerFlushIT extends TestSuiteBase implements
TestResource {
+
+ private static final String STARROCKS_IMAGE =
"seatunnelhub/starrocks-starter:2.2.1";
+ private static final String STARROCKS_HOST = "starrocks_timer_flush_e2e";
+ private static final int STARROCKS_QUERY_PORT = 9030;
+ private static final String MYSQL_HOST = "mysql_starrocks_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 SINK_DATABASE = "timer_flush";
+ private static final String SINK_TABLE = "starrocks_timer_flush";
+ private static final String JDBC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/Jdbc/lib";
+ 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<?> starRocksServer;
+ private Connection starRocksConnection;
+
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
+ container -> {
+ copyMySQLDriverToContainer(container, JDBC_PLUGIN_LIB);
+ copyMySQLDriverToContainer(container, MYSQL_CDC_PLUGIN_LIB);
+ };
+
+ @BeforeAll
+ @Override
+ public void startUp() {
+ starRocksServer =
+ new GenericContainer<>(STARROCKS_IMAGE)
+ .withNetwork(NETWORK)
+ .withNetworkAliases(STARROCKS_HOST)
+ .withExposedPorts(STARROCKS_QUERY_PORT)
+ .withLogConsumer(
+ new Slf4jLogConsumer(
+
DockerLoggerFactory.getLogger(STARROCKS_IMAGE)));
+ Startables.deepStart(Stream.of(starRocksServer,
MYSQL_CONTAINER)).join();
+ given().ignoreExceptions()
+ .await()
+ .atMost(360, TimeUnit.SECONDS)
+ .untilAsserted(this::initializeStarRocksConnection);
+ shopDatabase.createAndInitialize();
+ initializeTimerFlushTable();
+ }
+
+ @TestTemplate
+ public void testStarRocksTimerFlush(TestContainer testContainer) throws
Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return testContainer.executeJob(
+
"/mysqlcdc_to_starrocks_timer_flush.conf", jobId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ await().atMost(2, TimeUnit.MINUTES)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ "The streaming job terminated before
reaching RUNNING");
+ Assertions.assertEquals(
+ "RUNNING",
testContainer.getJobStatus(jobId));
+ });
+
+ await().atMost(120, TimeUnit.SECONDS)
+ .ignoreExceptions()
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ "The streaming job must still be
running when timer flush publishes the snapshot");
+ Assertions.assertEquals(9, tableCount());
+ });
+
+ 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(
+ () -> {
+ Assertions.assertFalse(
+ jobFuture.isDone(),
+ "The streaming job must still be
running when timer flush publishes the binlog event");
+ Assertions.assertEquals(10, tableCount());
+ });
+ } 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 void initializeStarRocksConnection() throws SQLException {
+ starRocksConnection =
+ DriverManager.getConnection(
+ String.format(
+ "jdbc:mysql://%s:%s",
+ starRocksServer.getHost(),
+
starRocksServer.getMappedPort(STARROCKS_QUERY_PORT)),
+ "root",
+ "");
+ }
+
+ private void initializeTimerFlushTable() {
+ String createTableSql =
+ "CREATE TABLE IF NOT EXISTS "
+ + SINK_DATABASE
+ + "."
+ + SINK_TABLE
+ + " ("
+ + "id INT NOT NULL, "
+ + "name VARCHAR(255), "
+ + "description VARCHAR(512), "
+ + "weight FLOAT) "
+ + "ENGINE=OLAP DUPLICATE KEY(id) "
+ + "DISTRIBUTED BY HASH(id) BUCKETS 1 "
+ + "PROPERTIES ('replication_num' = '1')";
+ try (Statement statement = starRocksConnection.createStatement()) {
+ statement.execute("CREATE DATABASE IF NOT EXISTS " +
SINK_DATABASE);
+ statement.execute(createTableSql);
+ statement.execute("TRUNCATE TABLE " + SINK_DATABASE + "." +
SINK_TABLE);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to initialize StarRocks timer
flush table", e);
+ }
+ }
+
+ private int tableCount() {
+ String sql = "SELECT COUNT(*) FROM " + SINK_DATABASE + "." +
SINK_TABLE;
+ try (Statement statement = starRocksConnection.createStatement();
+ ResultSet resultSet = statement.executeQuery(sql)) {
+ Assertions.assertTrue(resultSet.next());
+ return resultSet.getInt(1);
+ } catch (SQLException e) {
+ throw new RuntimeException("Failed to count rows in StarRocks
timer flush table", e);
+ }
+ }
+
+ private void copyMySQLDriverToContainer(GenericContainer<?> container,
String pluginLibDir)
+ 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 " +
pluginLibDir);
+ Assertions.assertEquals(0, extraCommands.getExitCode(),
extraCommands.getStderr());
+ container.copyFileToContainer(
+ MountableFile.forHostPath(driverJarPath),
+ pluginLibDir + "/" + 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() throws Exception {
+ if (starRocksConnection != null) {
+ starRocksConnection.close();
+ }
+ if (starRocksServer != null) {
+ starRocksServer.close();
+ }
+ MYSQL_CONTAINER.close();
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/mysqlcdc_to_starrocks_timer_flush.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/mysqlcdc_to_starrocks_timer_flush.conf
new file mode 100644
index 0000000000..077cb3c222
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-starrocks-e2e/src/test/resources/mysqlcdc_to_starrocks_timer_flush.conf
@@ -0,0 +1,53 @@
+#
+# 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 = 5670
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["shop.products"]
+ url = "jdbc:mysql://mysql_starrocks_timer_flush_e2e:3306/shop"
+ }
+}
+
+sink {
+ StarRocks {
+ nodeUrls = ["starrocks_timer_flush_e2e:8030"]
+ base-url = "jdbc:mysql://starrocks_timer_flush_e2e:9030/timer_flush"
+ username = root
+ password = ""
+ database = "timer_flush"
+ table = "starrocks_timer_flush"
+ labelPrefix = "timer-flush-"
+ batch_max_rows = 100000
+ batch_max_bytes = 104857600
+ schema_save_mode = "IGNORE"
+ data_save_mode = "APPEND_DATA"
+ starrocks.config = {
+ format = "JSON"
+ }
+ }
+}
