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 bd5a19a6c5 [Fix][Connector-V2] Flush Doris load before schema change
(#11156)
bd5a19a6c5 is described below
commit bd5a19a6c51b754e0c42c2e1fd9e319f5e41dcec
Author: QuakeWang <[email protected]>
AuthorDate: Wed Jul 8 21:11:11 2026 +0800
[Fix][Connector-V2] Flush Doris load before schema change (#11156)
Signed-off-by: QuakeWang <[email protected]>
---
.../introduction/configuration/schema-evolution.md | 2 +
.../introduction/configuration/schema-evolution.md | 2 +
.../doris/sink/writer/DorisSinkWriter.java | 43 +++++-
.../doris/sink/writer/DorisSinkWriterTest.java | 162 +++++++++++++++++++--
4 files changed, 195 insertions(+), 14 deletions(-)
diff --git a/docs/en/introduction/configuration/schema-evolution.md
b/docs/en/introduction/configuration/schema-evolution.md
index f889fce1b2..bc6637d714 100644
--- a/docs/en/introduction/configuration/schema-evolution.md
+++ b/docs/en/introduction/configuration/schema-evolution.md
@@ -254,6 +254,8 @@ sink {
}
```
+> **Note (schema evolution + 2PC):** With `sink.enable-2pc = "true"`, Doris
schema evolution only supports `format = "json"` because JSON loads match
columns by name. Positional formats such as CSV are rejected at runtime for
schema evolution with 2PC enabled. Use `format = "json"` or set
`sink.enable-2pc = "false"` so the sink can flush buffered rows before applying
the DDL.
+
### Mysql-CDC -> Jdbc-Postgres
```hocon
env {
diff --git a/docs/zh/introduction/configuration/schema-evolution.md
b/docs/zh/introduction/configuration/schema-evolution.md
index 4baf957d0b..f87654500d 100644
--- a/docs/zh/introduction/configuration/schema-evolution.md
+++ b/docs/zh/introduction/configuration/schema-evolution.md
@@ -255,6 +255,8 @@ sink {
}
```
+> **注意（schema 演进 + 2PC）：** 当 `sink.enable-2pc = "true"` 时，Doris schema 演进仅支持
`format = "json"`，因为 JSON load 会按列名匹配。CSV 等位置敏感格式在启用 2PC 的 schema
演进场景下会被运行时拒绝。请使用 `format = "json"`，或设置 `sink.enable-2pc = "false"`，让 sink 可以在应用
DDL 前先 flush 已缓冲的数据。
+
### Mysql-CDC -> Jdbc-Postgres
```hocon
env {
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 1ba86a87bb..3017ed9f3f 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
@@ -204,7 +204,21 @@ public class DorisSinkWriter
}
@Override
- public void applySchemaChange(SchemaChangeEvent event) {
+ public void applySchemaChange(SchemaChangeEvent event) throws IOException {
+ validateSchemaChangeCompatibility();
+
+ // The in-flight stream load may still buffer rows serialized with the
previous schema.
+ // In non-2PC mode each micro-batch is an independent load, so close
the current load
+ // first (committing the buffered rows against the still-unaltered
table) before applying
+ // the DDL, then reopen a fresh load so subsequent rows are loaded
against the new schema.
+ // Mixing schemas within a single load corrupts data (e.g. column
mismatch for csv/dropped
+ // columns). 2PC keeps a single transaction per checkpoint, so only
name-based JSON loads
+ // are allowed to cross a schema-change boundary.
+ boolean flushBeforeSchemaChange = !dorisSinkConfig.getEnable2PC();
+ if (flushBeforeSchemaChange) {
+ flush();
+ }
+
this.tableSchema = tableSchemaChanger.reset(tableSchema).apply(event);
SeaTunnelRowType seaTunnelRowType =
tableSchema.toPhysicalRowDataType();
this.serializer = createSerializer(this.dorisSinkConfig,
seaTunnelRowType);
@@ -213,7 +227,27 @@ public class DorisSinkWriter
schemaChangeManager.applySchemaChange(sinkTablePath, event);
} catch (Exception e) {
throw new DorisSchemaChangeException(
- DorisConnectorErrorCode.SCHEMA_CHANGE_FAILED, "Failed to
schemaChange");
+ DorisConnectorErrorCode.SCHEMA_CHANGE_FAILED, "Failed to
schemaChange", e);
+ }
+
+ if (flushBeforeSchemaChange) {
+ startLoad(labelGenerator.generateLabel(lastCheckpointId));
+ }
+ }
+
+ private void validateSchemaChangeCompatibility() {
+ if (!dorisSinkConfig.getEnable2PC()) {
+ return;
+ }
+ String format =
dorisSinkConfig.getStreamLoadProps().getProperty(LoadConstants.FORMAT_KEY);
+ if (!LoadConstants.JSON.equalsIgnoreCase(format)) {
+ throw new DorisSchemaChangeException(
+ DorisConnectorErrorCode.SCHEMA_CHANGE_FAILED,
+ String.format(
+ "Doris schema evolution with sink.enable-2pc=true
only supports "
+ + "stream load format=json, but current
format is %s. "
+ + "Use format=json or set
sink.enable-2pc=false.",
+ format));
}
}
@@ -254,6 +288,11 @@ public class DorisSinkWriter
this.dorisStreamLoad.startLoad(label);
}
+ // visible for testing: allows injecting a mocked SchemaChangeManager to
avoid real HTTP calls.
+ void setSchemaChangeManager(SchemaChangeManager schemaChangeManager) {
+ this.schemaChangeManager = schemaChangeManager;
+ }
+
@Override
public void abortPrepare() {
if (dorisSinkConfig.getEnable2PC()) {
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 7dc4d3bd4d..5cc77387ae 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
@@ -20,17 +20,23 @@ 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.table.catalog.CatalogTable;
+import org.apache.seatunnel.api.table.catalog.PhysicalColumn;
+import org.apache.seatunnel.api.table.catalog.TableIdentifier;
import org.apache.seatunnel.api.table.catalog.TablePath;
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.api.table.type.SeaTunnelRowType;
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;
+import org.apache.seatunnel.connectors.doris.schema.SchemaChangeManager;
import org.apache.seatunnel.connectors.doris.sink.LoadStatus;
import org.apache.seatunnel.connectors.doris.sink.committer.DorisCommitInfo;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
+import org.mockito.InOrder;
import java.util.ArrayList;
import java.util.Arrays;
@@ -40,11 +46,14 @@ import java.util.List;
import java.util.Map;
import java.util.Optional;
+import static org.mockito.ArgumentMatchers.any;
import static org.mockito.ArgumentMatchers.anyLong;
import static org.mockito.ArgumentMatchers.anyString;
import static org.mockito.Mockito.doNothing;
import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.inOrder;
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;
@@ -198,7 +207,130 @@ public class DorisSinkWriterTest {
verify(frontendLoad, times(1)).close();
}
+ @Test
+ void
testApplySchemaChangeFlushesInFlightLoadBeforeAlterForNonTwoPhaseCommit()
+ throws Exception {
+ DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
+ when(frontendLoad.stopLoad()).thenReturn(successRespContent(31L));
+ SchemaChangeManager schemaChangeManager =
mock(SchemaChangeManager.class);
+
+ RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory();
+ factory.register("fe1:8030", frontendLoad);
+
+ DorisSinkWriter writer = null;
+ try {
+ writer =
+ new DorisSinkWriter(
+ new DefaultSinkWriterContext(0, 1),
+ new ArrayList<>(),
+ mockCatalogTable(),
+ createSinkConfig(false, false),
+ "job_1",
+ factory);
+ writer.setSchemaChangeManager(schemaChangeManager);
+
+ writer.applySchemaChange(addColumnEvent());
+
+ // The crux of the fix is the ordering: the in-flight load is
closed (committing the
+ // old-schema rows against the still-unaltered table) BEFORE the
DDL is applied, and a
+ // fresh load is reopened only AFTER the DDL. Assert the order
across both mocks so a
+ // regression such as "DDL -> stopLoad" (which would commit
old-schema rows against the
+ // new table structure) is caught, not just the call counts.
+ InOrder inOrder = inOrder(frontendLoad, schemaChangeManager);
+ inOrder.verify(frontendLoad).stopLoad();
+ inOrder.verify(schemaChangeManager)
+ .applySchemaChange(any(TablePath.class),
any(SchemaChangeEvent.class));
+ inOrder.verify(frontendLoad).startLoad(anyString());
+ // exactly one flush here (close() would add another for non-2PC,
hence verify before
+ // close), and a fresh load reopened (init load + the reopened
one).
+ verify(frontendLoad, times(1)).stopLoad();
+ verify(frontendLoad, times(2)).startLoad(anyString());
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+
+ @Test
+ void testApplySchemaChangeKeepsTransactionIntactForTwoPhaseCommit() throws
Exception {
+ DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
+ doNothing().when(frontendLoad).abortPreCommit(anyString(), anyLong());
+ SchemaChangeManager schemaChangeManager =
mock(SchemaChangeManager.class);
+
+ RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory();
+ factory.register("fe1:8030", frontendLoad);
+
+ DorisSinkWriter writer = null;
+ try {
+ writer =
+ new DorisSinkWriter(
+ new DefaultSinkWriterContext(0, 1),
+ new ArrayList<>(),
+ mockCatalogTable(),
+ createSinkConfig(false, true),
+ "job_1",
+ factory);
+ writer.setSchemaChangeManager(schemaChangeManager);
+
+ writer.applySchemaChange(addColumnEvent());
+
+ // 2PC must not flush mid-checkpoint: no extra stopLoad/startLoad,
only the init load.
+ verify(frontendLoad, never()).stopLoad();
+ verify(frontendLoad, times(1)).startLoad(anyString());
+ verify(schemaChangeManager, times(1))
+ .applySchemaChange(any(TablePath.class),
any(SchemaChangeEvent.class));
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+
+ @Test
+ void testApplySchemaChangeRejectsTwoPhaseCommitWithCsvFormat() throws
Exception {
+ DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class);
+ doNothing().when(frontendLoad).abortPreCommit(anyString(), anyLong());
+ SchemaChangeManager schemaChangeManager =
mock(SchemaChangeManager.class);
+
+ RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory();
+ factory.register("fe1:8030", frontendLoad);
+
+ DorisSinkWriter writer = null;
+ try {
+ writer =
+ new DorisSinkWriter(
+ new DefaultSinkWriterContext(0, 1),
+ new ArrayList<>(),
+ mockCatalogTable(),
+ createSinkConfig(false, true, "csv"),
+ "job_1",
+ factory);
+ writer.setSchemaChangeManager(schemaChangeManager);
+
+ DorisSinkWriter schemaChangeWriter = writer;
+ DorisSchemaChangeException exception =
+ Assertions.assertThrows(
+ DorisSchemaChangeException.class,
+ () ->
schemaChangeWriter.applySchemaChange(addColumnEvent()));
+
Assertions.assertTrue(exception.getMessage().contains("format=json"));
+
+ verify(frontendLoad, never()).stopLoad();
+ verify(frontendLoad, times(1)).startLoad(anyString());
+ verify(schemaChangeManager, never())
+ .applySchemaChange(any(TablePath.class),
any(SchemaChangeEvent.class));
+ } finally {
+ if (writer != null) {
+ writer.close();
+ }
+ }
+ }
+
private DorisSinkConfig createSinkConfig(boolean directToBe, boolean
enable2PC) {
+ return createSinkConfig(directToBe, enable2PC, "json");
+ }
+
+ private DorisSinkConfig createSinkConfig(boolean directToBe, boolean
enable2PC, String format) {
Map<String, Object> options = new HashMap<>();
options.put("fenodes", "fe1:8030");
options.put("benodes", "be1:8040");
@@ -209,31 +341,37 @@ public class DorisSinkWriterTest {
options.put("database", "test_db");
options.put("table", "test_table");
options.put("sink.label-prefix", "test_job");
- options.put("doris.config", createStreamLoadProperties());
+ options.put("doris.config", createStreamLoadProperties(format));
return DorisSinkConfig.of(ReadonlyConfig.fromMap(options));
}
- private Map<String, String> createStreamLoadProperties() {
+ private Map<String, String> createStreamLoadProperties(String format) {
Map<String, String> properties = new HashMap<>();
- properties.put("format", "json");
+ properties.put("format", format);
properties.put("read_json_by_line", "true");
return properties;
}
private CatalogTable mockCatalogTable() {
+ TableSchema tableSchema =
+ TableSchema.builder()
+ .column(
+ PhysicalColumn.of(
+ "id", BasicType.LONG_TYPE, (Long)
null, true, null, null))
+ .build();
CatalogTable catalogTable = mock(CatalogTable.class);
when(catalogTable.getTablePath()).thenReturn(TablePath.of("test_db",
"test_table"));
-
when(catalogTable.getTableSchema()).thenReturn(mock(TableSchema.class));
- when(catalogTable.getSeaTunnelRowType())
- .thenReturn(
- new SeaTunnelRowType(
- new String[] {"id"},
- new
org.apache.seatunnel.api.table.type.SeaTunnelDataType[] {
- BasicType.LONG_TYPE
- }));
+ when(catalogTable.getTableSchema()).thenReturn(tableSchema);
+
when(catalogTable.getSeaTunnelRowType()).thenReturn(tableSchema.toPhysicalRowDataType());
return catalogTable;
}
+ private AlterTableAddColumnEvent addColumnEvent() {
+ return AlterTableAddColumnEvent.add(
+ TableIdentifier.of("doris", "test_db", null, "test_table"),
+ PhysicalColumn.of("age", BasicType.INT_TYPE, (Long) null,
true, null, null));
+ }
+
private RespContent successRespContent(long txnId) {
RespContent respContent = new RespContent();
respContent.setTxnId(txnId);
