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 3ba5470a3d [Feature][Connector-V2] Support timer flush for Doris sink
(#11506)
3ba5470a3d is described below
commit 3ba5470a3d581dd59b1af14c92d763f34edf3d30
Author: zhiwei.niu <[email protected]>
AuthorDate: Sat Jul 25 21:37:07 2026 +0800
[Feature][Connector-V2] Support timer flush for Doris sink (#11506)
---
docs/en/connectors/sink/Doris.md | 29 ++-
docs/zh/connectors/sink/Doris.md | 29 ++-
.../doris/sink/writer/DorisSinkWriter.java | 13 ++
.../doris/sink/writer/DorisSinkWriterTest.java | 66 +++++++
.../e2e/connector/doris/DorisTimerFlushIT.java | 200 +++++++++++++++++++++
.../resources/mysqlcdc_to_doris_timer_flush.conf | 53 ++++++
6 files changed, 388 insertions(+), 2 deletions(-)
diff --git a/docs/en/connectors/sink/Doris.md b/docs/en/connectors/sink/Doris.md
index 0ff0b67130..d70c3ac2ee 100644
--- a/docs/en/connectors/sink/Doris.md
+++ b/docs/en/connectors/sink/Doris.md
@@ -21,7 +21,7 @@ import ChangeLog from '../changelog/connector-doris.md';
- [x] [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
@@ -195,6 +195,33 @@ This is because the total amount of data arriving at the
end may not exceed the
Otherwise, if you enable the 2pc by the property `sink.enable-2pc=true`.The
`sink.buffer-size` will have no effect. So only the checkpoint can trigger the
commit.
+### Timer flush on Zeta
+
+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 finish the current Stream Load before `doris.batch.size` is
reached.
+
+Timer flush is registered only when `sink.enable-2pc=false`. It is
intentionally disabled when `sink.enable-2pc=true` because flushing and opening
a new Stream Load between checkpoints would break the 2PC transaction boundary
and exactly-once guarantee. The initial timer flush implementation therefore
provides at-least-once delivery only.
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ Doris {
+ fenodes = "doris-fe:8030"
+ username = root
+ password = ""
+ database = "mydb"
+ table = "mytable"
+ sink.label-prefix = "timer-flush"
+ sink.enable-2pc = false
+ doris.batch.size = 10000
+ }
+}
+```
+
## Troubleshooting 307 Temporary Redirect
If the job fails with `HTTP/1.1 307 Temporary Redirect`, check the following
items first:
diff --git a/docs/zh/connectors/sink/Doris.md b/docs/zh/connectors/sink/Doris.md
index 32ace9bbda..70af62651b 100644
--- a/docs/zh/connectors/sink/Doris.md
+++ b/docs/zh/connectors/sink/Doris.md
@@ -21,7 +21,7 @@ import ChangeLog from '../changelog/connector-doris.md';
- [x] [精确一次](../../introduction/concepts/connector-v2-features.md)
- [x] [变更数据捕获](../../introduction/concepts/connector-v2-features.md)
- [x] [支持多表写入](../../introduction/concepts/connector-v2-features.md)
-- [ ] [定时刷新](../../introduction/concepts/connector-v2-features.md)
+- [x] [定时刷新](../../introduction/concepts/connector-v2-features.md)
## 描述
@@ -192,6 +192,33 @@ CREATE TABLE IF NOT EXISTS `${database}`.`${table_name}`
此外，如果你通过`sink.enable-2pc=true`属性启用2pc。`sink.buffer-size`将会失去作用，只有检查点才能触发提交。
+### Zeta 定时刷新
+
+该引擎级能力仅由 Zeta 支持，Spark 和 Flink 不会注入 `FlushSignal`。在 Zeta 中，可以在 `env` 块配置
`sink.flush.interval`，使当前 Stream Load 在达到 `doris.batch.size` 之前也能定时完成。
+
+仅当 `sink.enable-2pc=false` 时才会注册定时刷新。`sink.enable-2pc=true` 时会明确禁用该能力，因为在
checkpoint 之间完成当前 Stream Load 并开启新 Stream Load 会破坏 2PC
事务边界和精确一次保证。因此，首版定时刷新仅提供至少一次语义。
+
+```hocon
+env {
+ job.mode = "STREAMING"
+ checkpoint.interval = 300000
+ sink.flush.interval = 5000
+}
+
+sink {
+ Doris {
+ fenodes = "doris-fe:8030"
+ username = root
+ password = ""
+ database = "mydb"
+ table = "mytable"
+ sink.label-prefix = "timer-flush"
+ sink.enable-2pc = false
+ doris.batch.size = 10000
+ }
+}
+```
+
## `307 Temporary Redirect` 排查
如果任务报错 `HTTP/1.1 307 Temporary Redirect`，建议优先检查以下项目：
diff --git
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
index 3017ed9f3f..c2cbf3a24d 100644
---
a/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
+++
b/seatunnel-connectors-v2/connector-doris/src/main/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriter.java
@@ -136,6 +136,9 @@ public class DorisSinkWriter
this.schemaChangeManager = new SchemaChangeManager(dorisSinkConfig);
this.streamLoadFactory = streamLoadFactory;
this.initializeLoad();
+ if (!dorisSinkConfig.getEnable2PC()) {
+ context.registerFlushAction(this::timerFlush);
+ }
}
private void initializeLoad() {
@@ -276,6 +279,16 @@ public class DorisSinkWriter
return respContent;
}
+ /**
+ * Finishes the current non-2PC Stream Load and opens a new one when the
Zeta engine delivers a
+ * timer flush signal.
+ */
+ private void timerFlush() throws IOException {
+ checkLoadException();
+ flush();
+ startLoad(labelGenerator.generateLabel(lastCheckpointId));
+ }
+
@Override
public List<DorisSinkState> snapshotState(long checkpointId) {
checkState(dorisStreamLoad != null);
diff --git
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriterTest.java
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriterTest.java
index 5cc77387ae..3ba8eda2ab 100644
---
a/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriterTest.java
+++
b/seatunnel-connectors-v2/connector-doris/src/test/java/org/apache/seatunnel/connectors/doris/sink/writer/DorisSinkWriterTest.java
@@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.doris.sink.writer;
import org.apache.seatunnel.api.configuration.ReadonlyConfig;
import org.apache.seatunnel.api.sink.DefaultSinkWriterContext;
+import org.apache.seatunnel.api.sink.SinkWriter;
import org.apache.seatunnel.api.table.catalog.CatalogTable;
import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableIdentifier;
@@ -27,6 +28,7 @@ import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.schema.event.AlterTableAddColumnEvent;
import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent;
import org.apache.seatunnel.api.table.type.BasicType;
+import org.apache.seatunnel.common.utils.function.RunnableWithException;
import org.apache.seatunnel.connectors.doris.config.DorisSinkConfig;
import
org.apache.seatunnel.connectors.doris.exception.DorisSchemaChangeException;
import org.apache.seatunnel.connectors.doris.rest.models.RespContent;
@@ -36,6 +38,7 @@ import
org.apache.seatunnel.connectors.doris.sink.committer.DorisCommitInfo;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
import org.mockito.InOrder;
import java.util.ArrayList;
@@ -60,6 +63,69 @@ import static org.mockito.Mockito.when;
public class DorisSinkWriterTest {
+ @Test
+ void testTimerFlushRegisteredAndReopensLoadForNonTwoPhaseCommit() throws
Exception {
+ DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
+ when(frontendLoad.stopLoad()).thenReturn(successRespContent(1L));
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+
+ RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory();
+ factory.register("fe1:8030", frontendLoad);
+
+ DorisSinkWriter writer = null;
+ try {
+ writer =
+ new DorisSinkWriter(
+ context,
+ new ArrayList<>(),
+ mockCatalogTable(),
+ createSinkConfig(false, false),
+ "job_1",
+ factory);
+
+ ArgumentCaptor<RunnableWithException> flushAction =
+ ArgumentCaptor.forClass(RunnableWithException.class);
+ verify(context).registerFlushAction(flushAction.capture());
+
+ flushAction.getValue().run();
+
+ verify(frontendLoad, times(1)).stopLoad();
+ verify(frontendLoad, times(2)).startLoad(anyString());
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+
+ @Test
+ void testTimerFlushNotRegisteredForTwoPhaseCommit() throws Exception {
+ DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
+ doNothing().when(frontendLoad).abortPreCommit(anyString(), anyLong());
+ SinkWriter.Context context = mock(SinkWriter.Context.class);
+
+ RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory();
+ factory.register("fe1:8030", frontendLoad);
+
+ DorisSinkWriter writer = null;
+ try {
+ writer =
+ new DorisSinkWriter(
+ context,
+ new ArrayList<>(),
+ mockCatalogTable(),
+ createSinkConfig(false, true),
+ "job_1",
+ factory);
+
+ verify(context, never()).registerFlushAction(any());
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+
@Test
void testInitializeLoadUsesFrontendWhenDirectToBeDisabled() throws
Exception {
DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisTimerFlushIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisTimerFlushIT.java
new file mode 100644
index 0000000000..dfa5c6e6f1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisTimerFlushIT.java
@@ -0,0 +1,200 @@
+/*
+ * 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.doris;
+
+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.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.output.Slf4jLogConsumer;
+import org.testcontainers.lifecycle.Startables;
+import org.testcontainers.utility.DockerLoggerFactory;
+
+import lombok.extern.slf4j.Slf4j;
+
+import java.sql.Connection;
+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;
+
+@Slf4j
+@DisabledOnContainer(
+ value = {},
+ type = {EngineType.SPARK, EngineType.FLINK},
+ disabledReason =
+ "engine-level timer flush (sink.flush.interval) is only
supported on Zeta engine")
+public class DorisTimerFlushIT extends AbstractDorisIT {
+
+ private static final String MYSQL_HOST = "mysql_doris_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 = "mysql_cdc";
+ private static final String MYSQL_CDC_PLUGIN_LIB =
"/tmp/seatunnel/plugins/MySQL-CDC/lib";
+ private static final String SINK_DATABASE = "timer_flush";
+ private static final String SINK_TABLE = "doris_timer_flush";
+ private static final MySqlContainer MYSQL_CONTAINER =
createMySqlContainer(MySqlVersion.V8_0);
+
+ private final UniqueDatabase inventoryDatabase =
+ new UniqueDatabase(MYSQL_CONTAINER, MYSQL_DATABASE);
+
+ @TestContainerExtension
+ private final ContainerExtendedFactory extendedFactory =
+ container -> copyMySQLDriverToContainer(container,
MYSQL_CDC_PLUGIN_LIB);
+
+ @BeforeAll
+ public void init() {
+ Startables.deepStart(Stream.of(MYSQL_CONTAINER)).join();
+ inventoryDatabase.createAndInitialize();
+ initializeTimerFlushTable();
+ }
+
+ @TestTemplate
+ public void testDorisTimerFlush(TestContainer testContainer) throws
Exception {
+ String jobId = String.valueOf(JobIdGenerator.newJobId());
+ CompletableFuture<Container.ExecResult> jobFuture =
+ CompletableFuture.supplyAsync(
+ () -> {
+ try {
+ return testContainer.executeJob(
+ "/mysqlcdc_to_doris_timer_flush.conf",
jobId);
+ } catch (Exception e) {
+ throw new RuntimeException(e);
+ }
+ });
+
+ try {
+ await().atMost(2, TimeUnit.MINUTES)
+ .pollInterval(2, TimeUnit.SECONDS)
+ .untilAsserted(
+ () -> {
+ if (jobFuture.isDone()) {
+ Container.ExecResult jobResult =
jobFuture.get();
+ Assertions.fail(
+ "The streaming job terminated
before reaching RUNNING: "
+ + jobResult.getStderr());
+ }
+ 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(2, tableCount());
+ });
+
+ try (Connection connection = inventoryDatabase.getJdbcConnection();
+ Statement statement = connection.createStatement()) {
+ statement.executeUpdate(
+ "INSERT INTO mysql_cdc_e2e_source_table (uuid, name,
score) "
+ + "VALUES (3, 'timer-flush', 100)");
+ }
+
+ 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(3, 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 initializeTimerFlushTable() {
+ String createTableSql =
+ "CREATE TABLE IF NOT EXISTS "
+ + SINK_DATABASE
+ + "."
+ + SINK_TABLE
+ + " ("
+ + "uuid BIGINT NOT NULL, "
+ + "name VARCHAR(128), "
+ + "score INT) "
+ + "DUPLICATE KEY(uuid) "
+ + "DISTRIBUTED BY HASH(uuid) BUCKETS 1 "
+ + "PROPERTIES ('replication_num' = '1')";
+ try (Statement statement = jdbcConnection.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 Doris timer flush
table", e);
+ }
+ }
+
+ private int tableCount() {
+ String sql = "SELECT COUNT(*) FROM " + SINK_DATABASE + "." +
SINK_TABLE;
+ try (Statement statement = jdbcConnection.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 Doris timer
flush table", 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
+ public void close() {
+ MYSQL_CONTAINER.close();
+ }
+}
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/mysqlcdc_to_doris_timer_flush.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/mysqlcdc_to_doris_timer_flush.conf
new file mode 100644
index 0000000000..830b0a36c1
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/mysqlcdc_to_doris_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 = 5664
+ username = "st_user_source"
+ password = "mysqlpw"
+ table-names = ["mysql_cdc.mysql_cdc_e2e_source_table"]
+ url = "jdbc:mysql://mysql_doris_timer_flush_e2e:3306/mysql_cdc"
+ }
+}
+
+sink {
+ Doris {
+ fenodes = "doris_e2e:8030"
+ username = root
+ password = ""
+ database = "timer_flush"
+ table = "doris_timer_flush"
+ sink.label-prefix = "timer-flush"
+ sink.enable-2pc = false
+ doris.batch.size = 100000
+ schema_save_mode = "IGNORE"
+ data_save_mode = "APPEND_DATA"
+ doris.config {
+ format = "json"
+ read_json_by_line = "true"
+ }
+ }
+}
