This is an automated email from the ASF dual-hosted git repository.
nzw921rx 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 91ba231cb4 [Fix][Connector-V2] Preserve MySQL DATETIME values across
time zones (#13025)
91ba231cb4 is described below
commit 91ba231cb4b524d42e22935d03f194d584cf2fd3
Author: Goutam Adwant <[email protected]>
AuthorDate: Tue Oct 27 22:19:18 2026 -0700
[Fix][Connector-V2] Preserve MySQL DATETIME values across time zones
(#13025)
---
docs/en/connectors/sink/Jdbc.md | 6 ++
.../introduction/concepts/incompatible-changes.md | 6 ++
docs/zh/connectors/sink/Jdbc.md | 6 ++
.../introduction/concepts/incompatible-changes.md | 6 ++
.../converter/AbstractJdbcRowConverter.java | 12 ++-
.../dialect/mysql/MysqlJdbcRowConverter.java | 7 ++
.../dialect/starrocks/StarRocksDialect.java | 20 ++++
.../dialect/mysql/MysqlJdbcRowConverterTest.java | 111 +++++++++++++++++++++
.../seatunnel/jdbc/JdbcMysqlTimestampIT.java | 92 +++++++++++++++++
.../resources/jdbc_mysql_datetime_round_trip.conf | 43 ++++++++
10 files changed, 307 insertions(+), 2 deletions(-)
diff --git a/docs/en/connectors/sink/Jdbc.md b/docs/en/connectors/sink/Jdbc.md
index f5a70fb80f..c7e48e81d5 100644
--- a/docs/en/connectors/sink/Jdbc.md
+++ b/docs/en/connectors/sink/Jdbc.md
@@ -8,6 +8,12 @@ The JDBC Sink connector writes SeaTunnel rows to databases
through a JDBC driver
If this is your first JDBC Sink job, start with [Choose a write
mode](#choose-a-write-mode) and the [Quick start](#quick-start-postgresql). The
complete option reference follows those sections.
+## MySQL temporal values
+
+MySQL `DATETIME` maps to SeaTunnel `TIMESTAMP` (`LocalDateTime`). The MySQL
JDBC sink binds its local date/time fields directly, preserving fractional
seconds supported by the target column even when the worker JVM and JDBC
connection time zones differ. MySQL `TIMESTAMP` maps to `TIMESTAMP_TZ` and
retains its instant-based handling.
+
+When upgrading from a version affected by
[#13024](https://github.com/apache/seatunnel/issues/13024), remove any manual
hour offset added to compensate for shifted `DATETIME` writes. Previously
stored values are not repaired automatically; compare them with the original
source before correcting or reloading data.
+
## Using Dependency
Install the `connector-jdbc` plugin first:
diff --git a/docs/en/introduction/concepts/incompatible-changes.md
b/docs/en/introduction/concepts/incompatible-changes.md
index 9b699028c5..a14cd53669 100644
--- a/docs/en/introduction/concepts/incompatible-changes.md
+++ b/docs/en/introduction/concepts/incompatible-changes.md
@@ -5,6 +5,12 @@ You need to check this document before you upgrade to related
version.
## dev
+### JDBC MySQL: DATETIME writes preserve local date/time fields
+
+The MySQL JDBC sink now binds SeaTunnel `TIMESTAMP` (`LocalDateTime`) directly
instead of converting it through `java.sql.Timestamp`. With different worker
JVM and JDBC connection time zones, affected versions could shift `DATETIME`
values or reject valid boundary values. Writes now preserve the local fields
and the fractional seconds supported by the target column. The `DATETIME`
source mapping and `TIMESTAMP_TZ` handling are unchanged.
+
+Remove any manual offset used to compensate for this defect. Existing stored
values are not changed automatically; verify them against the original source
before correcting or reloading them. No new configuration is required.
+
### Doris and StarRocks sources: a null MAP or ROW column no longer reads as
an empty map or fails the job
- **Behavior change: a null complex cell from an Arrow stream now reads as
`null`**
diff --git a/docs/zh/connectors/sink/Jdbc.md b/docs/zh/connectors/sink/Jdbc.md
index 8b85b2bf1a..a5d5e21c02 100644
--- a/docs/zh/connectors/sink/Jdbc.md
+++ b/docs/zh/connectors/sink/Jdbc.md
@@ -8,6 +8,12 @@ JDBC Sink 通过数据库厂商提供的 JDBC 驱动写入数据。它支持批
第一次配置 JDBC Sink 时，请先阅读[选择写入模式](#选择写入模式)和[快速入门](#快速入门postgresql)，再按需查阅后面的完整参数说明。
+## MySQL 时间值
+
+MySQL `DATETIME` 映射为 SeaTunnel `TIMESTAMP`（`LocalDateTime`）。MySQL JDBC Sink
直接绑定本地日期和时间字段，即使 Worker JVM 与 JDBC 连接使用不同的时区，也会保留目标列支持的小数秒精度。MySQL `TIMESTAMP`
映射为 `TIMESTAMP_TZ`，仍按时间点处理。
+
+从受 [#13024](https://github.com/apache/seatunnel/issues/13024) 影响的版本升级时，请移除为补偿
`DATETIME` 写入偏移而手动添加的小时偏移。已经写入的数据不会自动修复；修正或重新加载之前，请先与原始数据进行比较。
+
## 使用依赖
先安装 `connector-jdbc` 插件：
diff --git a/docs/zh/introduction/concepts/incompatible-changes.md
b/docs/zh/introduction/concepts/incompatible-changes.md
index 7a5ceb6b78..021601be5f 100644
--- a/docs/zh/introduction/concepts/incompatible-changes.md
+++ b/docs/zh/introduction/concepts/incompatible-changes.md
@@ -4,6 +4,12 @@
## dev
+### JDBC MySQL：DATETIME 写入保留本地日期和时间字段
+
+MySQL JDBC Sink 现在直接绑定 SeaTunnel `TIMESTAMP`（`LocalDateTime`），不再通过
`java.sql.Timestamp` 转换。当 Worker JVM 与 JDBC 连接的时区不同时，受影响的版本可能导致 `DATETIME`
值偏移，或拒绝合法的边界值。现在写入会保留本地字段及目标列支持的小数秒精度。`DATETIME` 的源端映射和 `TIMESTAMP_TZ` 的处理保持不变。
+
+请移除为补偿此缺陷而设置的手动偏移。已经存储的数据不会自动修改；修正或重新加载之前，请先与原始数据核对。无需新增配置。
+
### Doris 与 StarRocks Source：MAP 或 ROW 列为空值时不再读成空 Map 或导致作业失败
- **行为变更：Arrow 流中为空的复杂类型单元格现在读为 `null`**
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
index c2a3fb04ab..74c8a125c0 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/converter/AbstractJdbcRowConverter.java
@@ -303,8 +303,7 @@ public abstract class AbstractJdbcRowConverter implements
JdbcRowConverter {
writeTime(statement, statementIndex, (LocalTime) value);
break;
case TIMESTAMP:
- LocalDateTime localDateTime = (LocalDateTime) value;
- statement.setTimestamp(statementIndex,
Timestamp.valueOf(localDateTime));
+ writeTimestamp(statement, statementIndex, (LocalDateTime)
value);
break;
case TIMESTAMP_TZ:
OffsetDateTime offsetDateTime = (OffsetDateTime) value;
@@ -360,4 +359,13 @@ public abstract class AbstractJdbcRowConverter implements
JdbcRowConverter {
throws SQLException {
statement.setTime(index, java.sql.Time.valueOf(time));
}
+
+ /**
+ * Binds a local timestamp using the legacy JDBC representation. Dialects
may override this
+ * method when their driver supports binding the local date/time fields
directly.
+ */
+ protected void writeTimestamp(PreparedStatement statement, int index,
LocalDateTime value)
+ throws SQLException {
+ statement.setTimestamp(index, Timestamp.valueOf(value));
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverter.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverter.java
index a4fe03ca0d..d22ee4ce47 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverter.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverter.java
@@ -46,4 +46,11 @@ public class MysqlJdbcRowConverter extends
AbstractJdbcRowConverter {
statement.setTimestamp(
index,
java.sql.Timestamp.valueOf(LocalDateTime.of(LocalDate.now(), time)));
}
+
+ /** Binds DATETIME fields without interpreting them in the JVM default
time zone. */
+ @Override
+ protected void writeTimestamp(PreparedStatement statement, int index,
LocalDateTime value)
+ throws SQLException {
+ statement.setObject(index, value);
+ }
}
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/starrocks/StarRocksDialect.java
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/starrocks/StarRocksDialect.java
index dd060fd918..c0e39de965 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/starrocks/StarRocksDialect.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/starrocks/StarRocksDialect.java
@@ -17,7 +17,14 @@
package
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.starrocks;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql.MysqlDialect;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.mysql.MysqlJdbcRowConverter;
+
+import java.sql.PreparedStatement;
+import java.sql.SQLException;
+import java.sql.Timestamp;
+import java.time.LocalDateTime;
public class StarRocksDialect extends MysqlDialect {
@@ -27,6 +34,19 @@ public class StarRocksDialect extends MysqlDialect {
this.fieldIde = fieldIde;
}
+ /** Keeps the existing StarRocks binding while MySQL uses its
DATETIME-specific writer. */
+ @Override
+ public JdbcRowConverter getRowConverter() {
+ return new MysqlJdbcRowConverter() {
+ @Override
+ protected void writeTimestamp(
+ PreparedStatement statement, int index, LocalDateTime
value)
+ throws SQLException {
+ statement.setTimestamp(index, Timestamp.valueOf(value));
+ }
+ };
+ }
+
@Override
public String hashModForField(String fieldName, int mod) {
return "ABS(murmur_hash3_32(" + quoteIdentifier(fieldName) + ") % " +
mod + ")";
diff --git
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverterTest.java
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverterTest.java
index 15c65b4e03..cb94e8278c 100644
---
a/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverterTest.java
+++
b/seatunnel-connectors-v2/connector-jdbc/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/mysql/MysqlJdbcRowConverterTest.java
@@ -21,22 +21,133 @@ import
org.apache.seatunnel.api.table.catalog.PhysicalColumn;
import org.apache.seatunnel.api.table.catalog.TableSchema;
import org.apache.seatunnel.api.table.type.LocalTimeType;
import org.apache.seatunnel.api.table.type.SeaTunnelRow;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.exception.JdbcConnectorException;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.converter.JdbcRowConverter;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.GenericDialect;
+import
org.apache.seatunnel.connectors.seatunnel.jdbc.internal.dialect.starrocks.StarRocksDialect;
import org.junit.jupiter.api.Test;
import org.mockito.ArgumentCaptor;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
+import java.sql.SQLException;
import java.sql.Timestamp;
+import java.time.LocalDateTime;
import java.time.LocalTime;
+import java.time.OffsetDateTime;
+import java.util.Calendar;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
import static org.mockito.Mockito.verify;
import static org.mockito.Mockito.when;
public class MysqlJdbcRowConverterTest {
+ @Test
+ public void testTimestampReadAndWritePreserveLocalFields() throws
Exception {
+ for (LocalDateTime expected :
+ new LocalDateTime[] {
+ LocalDateTime.parse("2026-01-01T12:34:56.123456"),
+ LocalDateTime.parse("2026-03-08T02:30:00.123456"),
+ LocalDateTime.parse("2026-11-01T01:30:00.654321")
+ }) {
+ TableSchema schema =
timestampSchema(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ MysqlJdbcRowConverter converter = new MysqlJdbcRowConverter();
+ ResultSet resultSet = mock(ResultSet.class);
+ when(resultSet.getObject(1,
LocalDateTime.class)).thenReturn(expected);
+ SeaTunnelRow row = converter.toInternal(resultSet, schema);
+ assertEquals(expected, row.getField(0));
+
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter.toExternal(schema, schema, row, statement);
+ verify(statement).setObject(1, expected);
+ verify(statement, never()).setTimestamp(eq(1),
any(Timestamp.class));
+ }
+ }
+
+ @Test
+ public void testTimestampNullBindingIsUnchanged() throws Exception {
+ TableSchema schema =
timestampSchema(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ MysqlJdbcRowConverter converter = new MysqlJdbcRowConverter();
+ ResultSet resultSet = mock(ResultSet.class);
+ SeaTunnelRow row = converter.toInternal(resultSet, schema);
+ assertEquals(null, row.getField(0));
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter.toExternal(schema, row, statement);
+ verify(statement).setObject(1, null);
+ verify(statement, never()).setTimestamp(eq(1), any(Timestamp.class));
+ }
+
+ @Test
+ public void testTimestampBindingFailureIsNotRetriedWithTimestamp() throws
Exception {
+ TableSchema schema =
timestampSchema(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ LocalDateTime value =
LocalDateTime.parse("2026-01-01T12:34:56.123456");
+ PreparedStatement statement = mock(PreparedStatement.class);
+ SQLException failure = new SQLException("binding failed", "22007");
+ doThrow(failure).when(statement).setObject(1, value);
+ JdbcConnectorException actual =
+ assertThrows(
+ JdbcConnectorException.class,
+ () ->
+ new MysqlJdbcRowConverter()
+ .toExternal(
+ schema,
+ new SeaTunnelRow(new Object[]
{value}),
+ statement));
+ assertSame(failure, actual.getCause());
+ verify(statement, never()).setTimestamp(eq(1), any(Timestamp.class));
+ }
+
+ @Test
+ public void testDefaultAndStarRocksTimestampBindingsAreUnchanged() throws
Exception {
+ LocalDateTime value =
LocalDateTime.parse("2026-01-01T12:34:56.123456");
+ TableSchema schema =
timestampSchema(LocalTimeType.LOCAL_DATE_TIME_TYPE);
+ for (JdbcRowConverter converter :
+ new JdbcRowConverter[] {
+ new GenericDialect().getRowConverter(), new
StarRocksDialect().getRowConverter()
+ }) {
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter.toExternal(schema, new SeaTunnelRow(new Object[]
{value}), statement);
+ ArgumentCaptor<Timestamp> timestamp =
ArgumentCaptor.forClass(Timestamp.class);
+ verify(statement).setTimestamp(eq(1), timestamp.capture());
+ assertEquals(value, timestamp.getValue().toLocalDateTime());
+ verify(statement, never()).setObject(1, value);
+ }
+ }
+
+ @Test
+ public void testTimestampTzBindingAndUtcFallbackAreUnchanged() throws
Exception {
+ TableSchema schema =
timestampSchema(LocalTimeType.OFFSET_DATE_TIME_TYPE);
+ OffsetDateTime value =
OffsetDateTime.parse("2026-01-01T12:34:56.123456+08:00");
+ MysqlJdbcRowConverter converter = new MysqlJdbcRowConverter();
+ PreparedStatement statement = mock(PreparedStatement.class);
+ converter.toExternal(schema, new SeaTunnelRow(new Object[] {value}),
statement);
+ verify(statement).setObject(1, value);
+
+ PreparedStatement legacy = mock(PreparedStatement.class);
+ doThrow(new SQLException("unsupported")).when(legacy).setObject(1,
value);
+ converter.toExternal(schema, new SeaTunnelRow(new Object[] {value}),
legacy);
+ ArgumentCaptor<Timestamp> timestamp =
ArgumentCaptor.forClass(Timestamp.class);
+ ArgumentCaptor<Calendar> calendar =
ArgumentCaptor.forClass(Calendar.class);
+ verify(legacy).setTimestamp(eq(1), timestamp.capture(),
calendar.capture());
+ assertEquals(value.toInstant(), timestamp.getValue().toInstant());
+ assertEquals("UTC", calendar.getValue().getTimeZone().getID());
+ }
+
+ private TableSchema timestampSchema(LocalTimeType<?> type) {
+ return TableSchema.builder()
+
.column(PhysicalColumn.builder().name("timestamp_col").dataType(type).build())
+ .build();
+ }
+
@Test
public void testTimeReadAndWritePreserveMicroseconds() throws Exception {
LocalTime expected = LocalTime.parse("12:34:56.123456");
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlTimestampIT.java
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlTimestampIT.java
index dee614aca1..9bd1cfaff5 100644
---
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlTimestampIT.java
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/java/org/apache/seatunnel/connectors/seatunnel/jdbc/JdbcMysqlTimestampIT.java
@@ -42,7 +42,11 @@ import lombok.extern.slf4j.Slf4j;
import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
+import java.sql.ResultSet;
import java.sql.Statement;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.stream.Stream;
@@ -169,6 +173,70 @@ public class JdbcMysqlTimestampIT extends TestSuiteBase
implements TestResource
+ result.getStderr());
}
+ /** Checks stored values independently of JDBC temporal deserialization. */
+ @TestTemplate
+ public void testMysqlDatetimeRoundTrip(TestContainer container) throws
Exception {
+ for (String connectionZone : Arrays.asList("UTC", "Asia/Shanghai")) {
+ for (boolean serverPrepared : Arrays.asList(false, true)) {
+ try (Connection connection = mysqlConnection();
+ Statement statement = connection.createStatement()) {
+ statement.execute("TRUNCATE TABLE datetime_sink");
+ }
+ Container.ExecResult result =
+ container.executeJob(
+ "/jdbc_mysql_datetime_round_trip.conf",
+ Arrays.asList(
+ "connection_zone=" + connectionZone,
+ "server_prepared=" + serverPrepared));
+ Assertions.assertEquals(
+ 0,
+ result.getExitCode(),
+ "DATETIME round trip failed: "
+ + connectionZone
+ + ", serverPrepared="
+ + serverPrepared
+ + "\n"
+ + result.getStderr());
+ Assertions.assertEquals(
+ storedTemporalValues("datetime_source"),
+ storedTemporalValues("datetime_sink"),
+ "Stored temporal values changed: "
+ + connectionZone
+ + ", serverPrepared="
+ + serverPrepared);
+ }
+ }
+ }
+
+ private Connection mysqlConnection() throws Exception {
+ return DriverManager.getConnection(
+ String.format(
+
"jdbc:mysql://%s:%d/%s?useSSL=false&allowPublicKeyRetrieval=true&serverTimezone=UTC",
+ mysqlContainer.getHost(),
+ mysqlContainer.getFirstMappedPort(),
+ MYSQL_DATABASE),
+ MYSQL_USER,
+ MYSQL_PASSWORD);
+ }
+
+ private List<String> storedTemporalValues(String table) throws Exception {
+ List<String> values = new ArrayList<>();
+ try (Connection connection = mysqlConnection();
+ Statement statement = connection.createStatement();
+ ResultSet result =
+ statement.executeQuery(
+ "SELECT id, DATE_FORMAT(dt_col, '%Y-%m-%d
%H:%i:%s.%f'), "
+ + "UNIX_TIMESTAMP(ts_col) FROM "
+ + table
+ + " ORDER BY id")) {
+ while (result.next()) {
+ values.add(
+ result.getInt(1) + ":" + result.getString(2) + ":" +
result.getString(3));
+ }
+ }
+ return values;
+ }
+
//
-------------------------------------------------------------------------
// Helpers
//
-------------------------------------------------------------------------
@@ -188,6 +256,30 @@ public class JdbcMysqlTimestampIT extends TestSuiteBase
implements TestResource
+ " dt_col DATETIME,"
+ " ts_col TIMESTAMP NULL"
+ ")");
+ stmt.execute(
+ "CREATE TABLE IF NOT EXISTS datetime_source ("
+ + "id INT PRIMARY KEY, dt_col DATETIME(6), ts_col
TIMESTAMP(6) NULL)");
+ stmt.execute("CREATE TABLE IF NOT EXISTS datetime_sink LIKE
datetime_source");
+ stmt.execute("TRUNCATE TABLE datetime_source");
+ List<String> datetimes =
+ Arrays.asList(
+ "2026-01-01 12:34:56.123456",
+ "2025-12-31 23:59:59.999999",
+ "2026-01-01 00:00:00.000001",
+ "1969-12-31 23:59:59.123456",
+ "2026-03-08 02:30:00.123456",
+ "2026-11-01 01:30:00.654321",
+ "1000-01-01 00:00:00.000000",
+ "9999-12-31 23:59:59.999999");
+ for (int i = 0; i < datetimes.size(); i++) {
+ stmt.execute(
+ "INSERT INTO datetime_source VALUES ("
+ + i
+ + ", '"
+ + datetimes.get(i)
+ + "', '2026-01-01 12:34:56.123456')");
+ }
+ stmt.execute("INSERT INTO datetime_source VALUES (8, NULL, NULL)");
// Insert a fixed wall-clock value: 2026-01-01 00:00:00
// DATETIME stores it as-is (NTZ); TIMESTAMP stores UTC and
displays in session TZ.
stmt.execute(
diff --git
a/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_datetime_round_trip.conf
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_datetime_round_trip.conf
new file mode 100644
index 0000000000..c242b49f98
--- /dev/null
+++
b/seatunnel-e2e/seatunnel-connector-v2-e2e/connector-jdbc-e2e/connector-jdbc-e2e-part-1/src/test/resources/jdbc_mysql_datetime_round_trip.conf
@@ -0,0 +1,43 @@
+#
+# 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.
+#
+
+# Run with UTC and Asia/Shanghai worker JVMs to exercise both mismatch
directions.
+env {
+ parallelism = 1
+ job.mode = "BATCH"
+}
+
+source {
+ Jdbc {
+ url =
"jdbc:mysql://mysql_ts_e2e:3306/ts_test?useSSL=false&serverTimezone="${connection_zone}
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "root"
+ password = "Abc!@#135_seatunnel"
+ query = "SELECT id, dt_col, ts_col FROM datetime_source"
+ }
+}
+
+sink {
+ Jdbc {
+ url =
"jdbc:mysql://mysql_ts_e2e:3306/ts_test?useSSL=false&serverTimezone="${connection_zone}"&useServerPrepStmts="${server_prepared}"&rewriteBatchedStatements=true"
+ driver = "com.mysql.cj.jdbc.Driver"
+ user = "root"
+ password = "Abc!@#135_seatunnel"
+ query = "INSERT INTO datetime_sink (id, dt_col, ts_col) VALUES (?, ?, ?)"
+ batch_size = 3
+ }
+}
