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-12945-570eb976113567efc3cbaa8160dbf8c1cab3c2ac in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 5666e8d594ef09560a5d9e65ace4cc3748bdac2a Author: zhengxiang378928908-code <[email protected]> AuthorDate: Mon Oct 26 04:51:07 2026 +0000 [Improve][Connector-V2][Doris] Warn when TIMESTAMP_TZ columns use the JVM default zone (#12945) Co-authored-by: Claude Opus 5.5 <[email protected]> --- docs/en/connectors/sink/Doris.md | 2 +- docs/zh/connectors/sink/Doris.md | 2 +- .../doris/sink/writer/DorisSinkWriter.java | 81 +++++++ .../doris/sink/writer/DorisSinkWriterTest.java | 239 ++++++++++++++++++++- .../seatunnel/e2e/connector/doris/DorisIT.java | 89 ++++++++ .../fake_timestamp_tz_to_doris_zone_explicit.conf | 57 +++++ .../fake_timestamp_tz_to_doris_zone_unset.conf | 56 +++++ 7 files changed, 523 insertions(+), 3 deletions(-) diff --git a/docs/en/connectors/sink/Doris.md b/docs/en/connectors/sink/Doris.md index eab0e94383..958b35c10f 100644 --- a/docs/en/connectors/sink/Doris.md +++ b/docs/en/connectors/sink/Doris.md @@ -62,7 +62,7 @@ The internal implementation of Doris sink connector is cached and imported by st | doris.batch.size | int | No | 1024 | the batch size of the write to doris each http request, when the row reaches the size or checkpoint is executed, the data of cached will write to server. | | needs_unsupported_type_casting | boolean | No | false | Whether to enable the unsupported type casting, such as Decimal64 to Double | | case_sensitive | boolean | No | true | Whether to preserve the original case of table and column names. When set to false, table and column names will be converted to lowercase. | -| sink.datetime-timezone | string | No | JVM default | Session zone ID (a `java.time.ZoneId`-compatible ID such as `Asia/Shanghai` or `+08:00`) used to convert `TIMESTAMP_TZ` values to wall-clock strings for Doris `DATETIME` columns. When unset, the JVM default zone is used (legacy behavior). Set this to the Doris session zone to avoid JVM-default coupling. | +| sink.datetime-timezone | string | No | JVM default | Session zone ID (a `java.time.ZoneId`-compatible ID such as `Asia/Shanghai` or `+08:00`) used to convert `TIMESTAMP_TZ` values to wall-clock strings for Doris `DATETIME` columns. When unset, the JVM default zone is used (legacy behavior). Set this to the Doris session zone to avoid JVM-default coupling. When the option is unset and the table has `TIMESTAMP_TZ` columns, the writer logs a warning that na [...] | schema_save_mode | Enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST | the schema save mode, please refer to `schema_save_mode` below | | data_save_mode | Enum | no | APPEND_DATA | the data save mode, please refer to `data_save_mode` below | | save_mode_create_template | string | no | see below | see below | diff --git a/docs/zh/connectors/sink/Doris.md b/docs/zh/connectors/sink/Doris.md index b85c4feddc..5459370948 100644 --- a/docs/zh/connectors/sink/Doris.md +++ b/docs/zh/connectors/sink/Doris.md @@ -62,7 +62,7 @@ Doris Sink连接器的内部实现是通过stream load批量缓存和导入的 | doris.batch.size | int | No | 1024 | 每次http请求写入doris的批量大小，当row达到该大小或者执行checkpoint时，缓存的数据就会写入服务器。 | | needs_unsupported_type_casting | boolean | No | false | 是否启用不支持的类型转换，例如 Decimal64 到 Double。 | | case_sensitive | boolean | No | true | 是否保留表名和字段名的原始大小写。当设置为 false 时，表名和字段名将被转换为小写。 | -| sink.datetime-timezone | string | No | JVM 默认时区 | 用于将 `TIMESTAMP_TZ` 值转换为 Doris `DATETIME` 列的 wall-clock 字符串的会话时区 ID（兼容 `java.time.ZoneId`，例如 `Asia/Shanghai` 或 `+08:00`）。未设置时使用 JVM 默认时区（旧行为）。请将其设置为 Doris 会话时区，以避免依赖 JVM 默认时区。 | +| sink.datetime-timezone | string | No | JVM 默认时区 | 用于将 `TIMESTAMP_TZ` 值转换为 Doris `DATETIME` 列的 wall-clock 字符串的会话时区 ID（兼容 `java.time.ZoneId`，例如 `Asia/Shanghai` 或 `+08:00`）。未设置时使用 JVM 默认时区（旧行为）。请将其设置为 Doris 会话时区，以避免依赖 JVM 默认时区。未设置且表中有 `TIMESTAMP_TZ` 列时，writer 会打印一条警告，列出这些列和实际使用的 JVM 时区。 | | schema_save_mode | Enum | no | CREATE_SCHEMA_WHEN_NOT_EXIST | schema保存模式，请参考下面的`schema_save_mode` | | data_save_mode | Enum | no | APPEND_DATA | 数据保存模式，请参考下面的`data_save_mode`。 | | save_mode_create_template | string | no | see below | 见下文。 | 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 bee0bab11f..8abed59058 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 @@ -28,6 +28,9 @@ import org.apache.seatunnel.api.table.catalog.TableSchema; import org.apache.seatunnel.api.table.schema.event.RestoreTableSchemaEvent; import org.apache.seatunnel.api.table.schema.event.SchemaChangeEvent; import org.apache.seatunnel.api.table.schema.handler.TableSchemaChangeEventDispatcher; +import org.apache.seatunnel.api.table.type.ArrayType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.connectors.doris.config.DorisSinkConfig; @@ -46,6 +49,7 @@ import org.apache.seatunnel.connectors.doris.util.UnsupportedTypeConverterUtils; import lombok.extern.slf4j.Slf4j; import java.io.IOException; +import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -84,6 +88,10 @@ public class DorisSinkWriter private SchemaChangeManager schemaChangeManager; private final DorisStreamLoadFactory streamLoadFactory; private String controlHostPort; + // Set once this writer instance has logged the TIMESTAMP_TZ zone warning, and only then: a + // schema without such columns leaves it unset, so a later schema change can still warn. It is + // not shared across writers or kept in state, so a restored writer may warn again. + private boolean timestampTzZoneWarningLogged; public DorisSinkWriter( SinkWriter.Context context, @@ -369,9 +377,82 @@ public class DorisSinkWriter private DorisSerializer createSerializer( DorisSinkConfig dorisSinkConfig, SeaTunnelRowType seaTunnelRowType) { + if (!timestampTzZoneWarningLogged) { + Optional<String> warning = + timestampTzZoneWarning( + catalogTable.getTablePath(), + seaTunnelRowType, + dorisSinkConfig.getDatetimeTimezone(), + ZoneId.systemDefault()); + if (warning.isPresent()) { + log.warn(warning.get()); + timestampTzZoneWarningLogged = true; + } + } return SeaTunnelRowSerializerFactory.createSerializer(dorisSinkConfig, seaTunnelRowType); } + /** + * Doris has no timezone-aware datetime type, so TIMESTAMP_TZ values are written as wall-clock + * time in {@code sink.datetime-timezone}, or in the JVM default zone when it is unset. The JVM + * zone can differ from the zone expected in Doris without any error, so say which zone is used. + */ + static Optional<String> timestampTzZoneWarning( + TablePath tablePath, + SeaTunnelRowType rowType, + ZoneId configuredZone, + ZoneId jvmDefaultZone) { + if (configuredZone != null) { + return Optional.empty(); + } + List<String> columns = new ArrayList<>(); + for (int i = 0; i < rowType.getTotalFields(); i++) { + collectTimestampTzPaths(rowType.getFieldName(i), rowType.getFieldType(i), columns); + } + if (columns.isEmpty()) { + return Optional.empty(); + } + return Optional.of( + String.format( + "Doris sink table %s has TIMESTAMP_TZ columns %s and " + + "'sink.datetime-timezone' is not set. These values are written " + + "as wall-clock time in the JVM default zone %s. If Doris expects " + + "another zone, set 'sink.datetime-timezone'.", + tablePath, columns, jvmDefaultZone)); + } + + /** + * Collects the TIMESTAMP_TZ columns of a type, including nested ones, for the warning message + * only. Nested paths use {@code .field} for row fields, {@code []} for array elements and + * {@code .<key>} / {@code .<value>} for map keys and values; they are not configuration paths. + */ + private static void collectTimestampTzPaths( + String path, SeaTunnelDataType<?> type, List<String> paths) { + switch (type.getSqlType()) { + case TIMESTAMP_TZ: + paths.add(path); + break; + case ROW: + SeaTunnelRowType rowType = (SeaTunnelRowType) type; + for (int i = 0; i < rowType.getTotalFields(); i++) { + collectTimestampTzPaths( + path + "." + rowType.getFieldName(i), rowType.getFieldType(i), paths); + } + break; + case ARRAY: + collectTimestampTzPaths( + path + "[]", ((ArrayType<?, ?>) type).getElementType(), paths); + break; + case MAP: + MapType<?, ?> mapType = (MapType<?, ?>) type; + collectTimestampTzPaths(path + ".<key>", mapType.getKeyType(), paths); + collectTimestampTzPaths(path + ".<value>", mapType.getValueType(), paths); + break; + default: + break; + } + } + private InitializedStreamLoad initializeStreamLoad( List<String> nodes, String nodeType, boolean abortPreCommitOnInit) { int nodeCount = nodes.size(); 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 3ba8eda2ab..236c923b79 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 @@ -27,7 +27,12 @@ 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.ArrayType; import org.apache.seatunnel.api.table.type.BasicType; +import org.apache.seatunnel.api.table.type.LocalTimeType; +import org.apache.seatunnel.api.table.type.MapType; +import org.apache.seatunnel.api.table.type.SeaTunnelDataType; +import org.apache.seatunnel.api.table.type.SeaTunnelRowType; import org.apache.seatunnel.common.utils.function.RunnableWithException; import org.apache.seatunnel.connectors.doris.config.DorisSinkConfig; import org.apache.seatunnel.connectors.doris.exception.DorisSchemaChangeException; @@ -36,11 +41,19 @@ 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.apache.logging.log4j.Level; +import org.apache.logging.log4j.LogManager; +import org.apache.logging.log4j.core.LogEvent; +import org.apache.logging.log4j.core.Logger; +import org.apache.logging.log4j.core.appender.AbstractAppender; +import org.apache.logging.log4j.core.config.Property; + import org.junit.jupiter.api.Assertions; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; import org.mockito.InOrder; +import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; import java.util.Collections; @@ -397,7 +410,15 @@ public class DorisSinkWriterTest { } private DorisSinkConfig createSinkConfig(boolean directToBe, boolean enable2PC, String format) { - Map<String, Object> options = new HashMap<>(); + return createSinkConfig(directToBe, enable2PC, format, Collections.emptyMap()); + } + + private DorisSinkConfig createSinkConfig( + boolean directToBe, + boolean enable2PC, + String format, + Map<String, Object> extraOptions) { + Map<String, Object> options = new HashMap<>(extraOptions); options.put("fenodes", "fe1:8030"); options.put("benodes", "be1:8040"); options.put("direct_to_be", directToBe); @@ -480,4 +501,220 @@ public class DorisSinkWriterTest { return createdHosts; } } + + @Test + void testTimestampTzZoneWarningNamesColumnsAndJvmZoneWhenZoneUnset() { + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "created_at", "updated_at", "local_ts"}, + new SeaTunnelDataType<?>[] { + BasicType.LONG_TYPE, + LocalTimeType.OFFSET_DATE_TIME_TYPE, + LocalTimeType.OFFSET_DATE_TIME_TYPE, + LocalTimeType.LOCAL_DATE_TIME_TYPE + }); + + Optional<String> warning = + DorisSinkWriter.timestampTzZoneWarning( + TablePath.of("db", "orders"), rowType, null, ZoneId.of("UTC")); + + Assertions.assertTrue(warning.isPresent()); + Assertions.assertTrue(warning.get().contains("db.orders"), warning.get()); + Assertions.assertTrue(warning.get().contains("[created_at, updated_at]"), warning.get()); + Assertions.assertTrue(warning.get().contains("JVM default zone UTC"), warning.get()); + Assertions.assertFalse(warning.get().contains("local_ts"), warning.get()); + } + + @Test + void testNoTimestampTzZoneWarningWhenZoneSetOrNoTimestampTzColumn() { + SeaTunnelRowType withTz = + new SeaTunnelRowType( + new String[] {"created_at"}, + new SeaTunnelDataType<?>[] {LocalTimeType.OFFSET_DATE_TIME_TYPE}); + SeaTunnelRowType withoutTz = + new SeaTunnelRowType( + new String[] {"id", "local_ts"}, + new SeaTunnelDataType<?>[] { + BasicType.LONG_TYPE, LocalTimeType.LOCAL_DATE_TIME_TYPE + }); + TablePath tablePath = TablePath.of("db", "orders"); + + Assertions.assertFalse( + DorisSinkWriter.timestampTzZoneWarning( + tablePath, withTz, ZoneId.of("Asia/Shanghai"), ZoneId.of("UTC")) + .isPresent()); + Assertions.assertFalse( + DorisSinkWriter.timestampTzZoneWarning(tablePath, withoutTz, null, ZoneId.of("UTC")) + .isPresent()); + } + + @Test + void testTimestampTzZoneWarningFindsNestedColumns() { + SeaTunnelRowType payload = + new SeaTunnelRowType( + new String[] {"name", "created_at"}, + new SeaTunnelDataType<?>[] { + BasicType.STRING_TYPE, LocalTimeType.OFFSET_DATE_TIME_TYPE + }); + SeaTunnelRowType rowType = + new SeaTunnelRowType( + new String[] {"id", "payload", "events", "attrs"}, + new SeaTunnelDataType<?>[] { + BasicType.LONG_TYPE, + payload, + ArrayType.OFFSET_DATE_TIME_ARRAY_TYPE, + new MapType<>( + BasicType.STRING_TYPE, LocalTimeType.OFFSET_DATE_TIME_TYPE) + }); + + Optional<String> warning = + DorisSinkWriter.timestampTzZoneWarning( + TablePath.of("db", "orders"), rowType, null, ZoneId.of("UTC")); + + Assertions.assertTrue(warning.isPresent()); + Assertions.assertTrue( + warning.get().contains("[payload.created_at, events[], attrs.<value>]"), + warning.get()); + } + + @Test + void testTimestampTzZoneWarningLoggedOnceWhenWriterCreatedWithTimestampTzColumn() + throws Exception { + List<String> warnings = + captureTimestampTzZoneWarnings( + () -> { + DorisSinkWriter writer = + new DorisSinkWriter( + new DefaultSinkWriterContext(0, 1), + new ArrayList<>(), + mockCatalogTable(timestampTzColumn("created_at")), + createSinkConfig(false, false), + "job_1", + frontendOnlyFactory()); + try { + writer.setSchemaChangeManager(mock(SchemaChangeManager.class)); + writer.applySchemaChange(addTimestampTzColumnEvent("updated_at")); + } finally { + writer.close(); + } + }); + + Assertions.assertEquals(1, warnings.size(), warnings.toString()); + Assertions.assertTrue( + warnings.get(0) + .contains("test_db.test_table has TIMESTAMP_TZ columns [created_at]"), + warnings.get(0)); + } + + @Test + void testTimestampTzZoneWarningLoggedOnceWhenSchemaChangeAddsTimestampTzColumn() + throws Exception { + List<String> warnings = + captureTimestampTzZoneWarnings( + () -> { + DorisSinkWriter writer = + new DorisSinkWriter( + new DefaultSinkWriterContext(0, 1), + new ArrayList<>(), + mockCatalogTable(), + createSinkConfig(false, false), + "job_1", + frontendOnlyFactory()); + try { + writer.setSchemaChangeManager(mock(SchemaChangeManager.class)); + writer.applySchemaChange(addColumnEvent()); + writer.applySchemaChange(addTimestampTzColumnEvent("created_at")); + writer.applySchemaChange(addTimestampTzColumnEvent("updated_at")); + } finally { + writer.close(); + } + }); + + Assertions.assertEquals(1, warnings.size(), warnings.toString()); + Assertions.assertTrue( + warnings.get(0).contains("TIMESTAMP_TZ columns [created_at]"), warnings.get(0)); + } + + @Test + void testNoTimestampTzZoneWarningLoggedWhenZoneConfigured() throws Exception { + Map<String, Object> options = new HashMap<>(); + options.put("sink.datetime-timezone", "Asia/Shanghai"); + List<String> warnings = + captureTimestampTzZoneWarnings( + () -> + new DorisSinkWriter( + new DefaultSinkWriterContext(0, 1), + new ArrayList<>(), + mockCatalogTable(timestampTzColumn("created_at")), + createSinkConfig(false, false, "json", options), + "job_1", + frontendOnlyFactory()) + .close()); + + Assertions.assertTrue(warnings.isEmpty(), warnings.toString()); + } + + private List<String> captureTimestampTzZoneWarnings(RunnableWithException action) + throws Exception { + List<String> warnings = Collections.synchronizedList(new ArrayList<>()); + AbstractAppender appender = + new AbstractAppender( + "timestamp-tz-zone-warning", null, null, true, Property.EMPTY_ARRAY) { + @Override + public void append(LogEvent event) { + String message = event.getMessage().getFormattedMessage(); + if (event.getLevel() == Level.WARN + && message.contains("'sink.datetime-timezone' is not set")) { + warnings.add(message); + } + } + }; + appender.start(); + Logger logger = (Logger) LogManager.getLogger(DorisSinkWriter.class); + Level originalLevel = logger.getLevel(); + logger.addAppender(appender); + logger.setLevel(Level.WARN); + try { + action.run(); + } finally { + logger.removeAppender(appender); + logger.setLevel(originalLevel); + appender.stop(); + } + return new ArrayList<>(warnings); + } + + private PhysicalColumn timestampTzColumn(String name) { + return PhysicalColumn.of( + name, LocalTimeType.OFFSET_DATE_TIME_TYPE, (Long) null, true, null, null); + } + + private AlterTableAddColumnEvent addTimestampTzColumnEvent(String name) { + return AlterTableAddColumnEvent.add( + TableIdentifier.of("doris", "test_db", null, "test_table"), + timestampTzColumn(name)); + } + + private RecordingStreamLoadFactory frontendOnlyFactory() throws Exception { + DorisStreamLoad frontendLoad = mock(DorisStreamLoad.class); + when(frontendLoad.stopLoad()).thenReturn(successRespContent(41L)); + RecordingStreamLoadFactory factory = new RecordingStreamLoadFactory(); + factory.register("fe1:8030", frontendLoad); + return factory; + } + + private CatalogTable mockCatalogTable(PhysicalColumn extraColumn) { + TableSchema tableSchema = + TableSchema.builder() + .column( + PhysicalColumn.of( + "id", BasicType.LONG_TYPE, (Long) null, true, null, null)) + .column(extraColumn) + .build(); + CatalogTable catalogTable = mock(CatalogTable.class); + when(catalogTable.getTablePath()).thenReturn(TablePath.of("test_db", "test_table")); + when(catalogTable.getTableSchema()).thenReturn(tableSchema); + when(catalogTable.getSeaTunnelRowType()).thenReturn(tableSchema.toPhysicalRowDataType()); + return catalogTable; + } } diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java index 8cf7d69bba..84226f0343 100644 --- a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/java/org/apache/seatunnel/e2e/connector/doris/DorisIT.java @@ -46,6 +46,9 @@ import java.sql.PreparedStatement; import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Statement; +import java.time.LocalDateTime; +import java.time.OffsetDateTime; +import java.time.ZoneId; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -67,6 +70,22 @@ public class DorisIT extends AbstractDorisIT { private static final String DUPLICATE_TABLE = "doris_duplicate_table"; private static final String sourceDB = "e2e_source"; private static final String sinkDB = "e2e_sink"; + private static final String TZ_UNSET_TABLE = "doris_tz_unset"; + private static final String TZ_EXPLICIT_TABLE = "doris_tz_explicit"; + private static final OffsetDateTime TZ_SOURCE_VALUE = + OffsetDateTime.parse("2025-09-28T02:54:08+08:00"); + private static final Pattern TIMESTAMP_TZ_WARNING = + Pattern.compile( + Pattern.quote( + "Doris sink table " + + sinkDB + + "." + + TZ_UNSET_TABLE + + " has TIMESTAMP_TZ columns [F_TS] and" + + " 'sink.datetime-timezone' is not set. These values" + + " are written as wall-clock time in the JVM default" + + " zone ") + + "(\\S+)\\. "); private Connection conn; private Map<String, String> checkColumnTypeMap = null; @@ -243,6 +262,76 @@ public class DorisIT extends AbstractDorisIT { checkSinkData(); } + @TestTemplate + public void testTimestampTzZoneWarning(TestContainer container) + throws IOException, InterruptedException { + initializeJdbcTable(); + try { + Container.ExecResult unsetResult = + container.executeJob("/fake_timestamp_tz_to_doris_zone_unset.conf"); + Assertions.assertEquals(0, unsetResult.getExitCode(), unsetResult.getStderr()); + Container.ExecResult explicitResult = + container.executeJob("/fake_timestamp_tz_to_doris_zone_explicit.conf"); + Assertions.assertEquals(0, explicitResult.getExitCode(), explicitResult.getStderr()); + + // Where the writer's log ends up depends on the engine, so search both. + String logs = + unsetResult.getStdout() + + unsetResult.getStderr() + + explicitResult.getStdout() + + explicitResult.getStderr() + + container.getServerLogs(); + Matcher warning = TIMESTAMP_TZ_WARNING.matcher(logs); + Assertions.assertTrue( + warning.find(), "No TIMESTAMP_TZ zone warning for " + TZ_UNSET_TABLE); + Assertions.assertFalse( + logs.contains("Doris sink table " + sinkDB + "." + TZ_EXPLICIT_TABLE), + "Unexpected TIMESTAMP_TZ zone warning for " + TZ_EXPLICIT_TABLE); + + // The warning names the zone the value was written in; the explicit zone is a control. + ZoneId jvmZone = ZoneId.of(warning.group(1)); + LocalDateTime expectedUnset = + TZ_SOURCE_VALUE.atZoneSameInstant(jvmZone).toLocalDateTime(); + LocalDateTime expectedExplicit = + TZ_SOURCE_VALUE.atZoneSameInstant(ZoneId.of("Asia/Shanghai")).toLocalDateTime(); + await().atMost(60, TimeUnit.SECONDS) + .pollInterval(2, TimeUnit.SECONDS) + .ignoreExceptions() + .untilAsserted( + () -> { + Assertions.assertEquals( + expectedUnset, readTimestampTz(TZ_UNSET_TABLE)); + Assertions.assertEquals( + expectedExplicit, readTimestampTz(TZ_EXPLICIT_TABLE)); + }); + } finally { + dropTable(sinkDB, TZ_UNSET_TABLE); + dropTable(sinkDB, TZ_EXPLICIT_TABLE); + } + } + + private LocalDateTime readTimestampTz(String table) throws SQLException { + try (Statement statement = conn.createStatement(); + ResultSet resultSet = + statement.executeQuery( + String.format( + "select cast(F_TS as string) from %s.%s where F_ID = 1", + sinkDB, table))) { + Assertions.assertTrue(resultSet.next(), "No row in " + table); + String value = resultSet.getString(1); + Assertions.assertFalse(resultSet.next(), "More than one row in " + table); + return LocalDateTime.parse(value.substring(0, 19).replace(' ', 'T')); + } + } + + private void dropTable(String db, String table) { + try (Statement statement = conn.createStatement()) { + statement.execute(String.format("DROP TABLE IF EXISTS %s.%s", db, table)); + } catch (SQLException e) { + throw new RuntimeException("Failed to drop " + db + "." + table, e); + } + } + private void checkAllTypeSinkData() { try { assertHasData(sourceDB, DUPLICATE_TABLE); diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_explicit.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_explicit.conf new file mode 100644 index 0000000000..3a4cedb58c --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_explicit.conf @@ -0,0 +1,57 @@ +# +# 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 = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + F_ID = "bigint" + F_TS = "timestamp_tz" + } + } + rows = [ + { + kind = INSERT + fields = [1, "2025-09-28T02:54:08+08:00"] + } + ] + } +} + +sink { + Doris { + fenodes = "doris_e2e:8030" + schema_save_mode = "RECREATE_SCHEMA" + username = root + password = "" + table.identifier = "e2e_sink.doris_tz_explicit" + sink.datetime-timezone = "Asia/Shanghai" + sink.enable-2pc = "false" + sink.label-prefix = "test_tz_explicit" + doris.config = { + format = "json" + read_json_by_line = "true" + } + save_mode_create_template = """CREATE TABLE IF NOT EXISTS `${database}`.`${table}` (${rowtype_fields}) ENGINE=OLAP UNIQUE KEY (`F_ID`) DISTRIBUTED BY HASH (`F_ID`) PROPERTIES ("replication_allocation" = "tag.location.default: 1")""" + } +} diff --git a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_unset.conf b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_unset.conf new file mode 100644 index 0000000000..890346f07b --- /dev/null +++ b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-doris-e2e/src/test/resources/fake_timestamp_tz_to_doris_zone_unset.conf @@ -0,0 +1,56 @@ +# +# 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 = "BATCH" +} + +source { + FakeSource { + schema = { + fields { + F_ID = "bigint" + F_TS = "timestamp_tz" + } + } + rows = [ + { + kind = INSERT + fields = [1, "2025-09-28T02:54:08+08:00"] + } + ] + } +} + +sink { + Doris { + fenodes = "doris_e2e:8030" + schema_save_mode = "RECREATE_SCHEMA" + username = root + password = "" + table.identifier = "e2e_sink.doris_tz_unset" + sink.enable-2pc = "false" + sink.label-prefix = "test_tz_unset" + doris.config = { + format = "json" + read_json_by_line = "true" + } + save_mode_create_template = """CREATE TABLE IF NOT EXISTS `${database}`.`${table}` (${rowtype_fields}) ENGINE=OLAP UNIQUE KEY (`F_ID`) DISTRIBUTED BY HASH (`F_ID`) PROPERTIES ("replication_allocation" = "tag.location.default: 1")""" + } +}
