[Feature][Connector-V2][Jdbc] MySQL support read TIMESTAMP_TZ with server_time_zone

[yzeng1618]

#9785

Purpose of this pull request

• Add the server_time_zone option to the JDBC source/sink, and thread this option through JdbcConnectionConfig, SimpleJdbcConnectionProvider, and `MysqlDialect. This ensures that the MySQL JDBC driver and row converter use a consistent session time zone, thereby avoiding the 8-hour time difference issue.

• Add TIMESTAMP_TZ read support for MySQL: SeaTunnel TIMESTAMP_TZ is read as OffsetDateTime using the configured server_time_zone (or system default as fallback).

Does this PR introduce any user-facing change?

Yes, but scope is limited:

• MySQL DATETIME mapping is kept as TIMESTAMP (LOCAL_DATE_TIME_TYPE) for backward compatibility.

• MySQL TIMESTAMP columns are now exposed as TIMESTAMP_TZ (OFFSET_DATE_TIME_TYPE) instead of TIMESTAMP to align with timezone-aware semantics and the new server_time_zone option.

• New configuration option server_time_zone is added to JDBC source and sink; when set, it controls both JDBC driver conversion and TIMESTAMP_TZ conversion in the MySQL row converter.

How was this patch tested?

Unit tests:

• Updated MySqlTypeConverterTest to cover DATETIME/TIMESTAMP/TIMESTAMP_TZ mappings.

• Added MysqlJdbcRowConverterTest for TIMESTAMP_TZ reading with system default zone.

• Added SimpleJdbcConnectionProviderTest to verify server_time_zone is correctly translated to the MySQL driver serverTimezone property and not overridden when user sets it explicitly.

E2E:

• Added jdbc_mysql_source_and_sink_server_time_zone.conf and JdbcMysqlIT uses it to validate server_time_zone through real MySQL containers (TIMESTAMP/TIMESTAMP_TZ round trip).

Check list

• If any new Jar binary package adding in your PR, please add License Notice according
  New License Guide
• [ * ] If necessary, please update the documentation to describe the new feature. https://github.com/apache/seatunnel/tree/dev/docs
• If you are contributing the connector code, please check that the following files are updated:
  1. Update plugin-mapping.properties and add new connector information in it
  2. Update the pom file of seatunnel-dist
  3. Add ci label in label-scope-conf
  4. Add e2e testcase in seatunnel-e2e
  5. Update connector plugin_config

[Copilot]

Pull request overview

This PR adds support for SeaTunnel's TIMESTAMP_TZ (timezone-aware timestamp) type when working with MySQL and other databases. The key change is that MySQL DATETIME columns are now mapped to OFFSET_DATE_TIME_TYPE (using system default timezone) to enable timezone-aware timestamp handling, while MySQL TIMESTAMP columns retain their legacy LOCAL_DATE_TIME_TYPE mapping for backward compatibility.

Key Changes:

• MySQL DATETIME columns now map to OFFSET_DATE_TIME_TYPE instead of LOCAL_DATE_TIME_TYPE (breaking change)
• Added MysqlJdbcRowConverter with custom toInternal() to handle TIMESTAMP_TZ reading using system default timezone
• Extended type converters across multiple connectors (StarRocks, SQL Server, PostgreSQL, Iceberg, Hudi, Elasticsearch, CDC) to handle TIMESTAMP_TZ
• Added E2E tests with new c_timestamp_tz column to validate the functionality

Reviewed changes

Copilot reviewed 28 out of 28 changed files in this pull request and generated 4 comments.

Show a summary per file

File	Description
MySqlTypeConverter.java	Changed DATETIME to map to OFFSET_DATE_TIME_TYPE; added TIMESTAMP_TZ reconversion support
MysqlJdbcRowConverter.java	Added custom toInternal() method to read TIMESTAMP_TZ using system default timezone
MySqlTypeConverterTest.java	Updated tests to expect OFFSET_DATE_TIME_TYPE for DATETIME; added TIMESTAMP_TZ reconversion tests
MysqlJdbcRowConverterTest.java	New unit tests for TIMESTAMP_TZ null handling and timezone conversion
JdbcMysqlIT.java, JdbcMysqlSplitIT.java, etc.	Added c_timestamp_tz column to test schemas and data
jdbc_mysql_*.conf	Updated INSERT statements to include c_timestamp_tz with reformatted placeholders
jdbc_mysql_*.sql	Added c_timestamp_tz to SELECT statements
StarRocksBaseSerializer.java	Added TIMESTAMP_TZ handling by converting to LocalDateTime
StarRocksTypeConverter.java, StarRocksSaveModeUtil.java, StarRocksDataTypeConvertor.java	Added TIMESTAMP_TZ case statements mapping to DATETIME/TIMESTAMP
SqlServerTypeConverter.java	Added TIMESTAMP_TZ reconversion to DATETIMEOFFSET (proper timezone support)
SqlserverJdbcRowConverter.java	Added TIMESTAMP_TZ write handling via toExternal()
PostgresTypeConverter.java	Added TIMESTAMP_TZ reconversion (maps to timestamp without timezone)
IcebergTypeMapper.java	Added TIMESTAMP_TZ mapping to TimestampType.withZone()
AvroSchemaConverter.java, RowDataToAvroConverters.java	Added TIMESTAMP_TZ Avro conversion support
ElasticSearchTypeConverter.java, KeyExtractor.java	Added TIMESTAMP_TZ handling for Elasticsearch
DefaultDataConverter.java (TiDB CDC)	Added convertToTimestampTz() using system default zone
SeaTunnelRowDebeziumDeserializationConverters.java	Added TIMESTAMP_TZ converter for Debezium CDC

Comments suppressed due to low confidence (1)

seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/internal/dialect/psql/PostgresTypeConverter.java:446

• PostgreSQL TIMESTAMP_TZ reconversion issue: When writing TIMESTAMP_TZ data to PostgreSQL, it gets converted to timestamp (without timezone) rather than timestamptz (with timezone). This means timezone information is lost when writing to PostgreSQL.

PostgreSQL natively supports timestamptz type which would preserve timezone information. Consider either:

1. Mapping TIMESTAMP_TZ to PostgreSQL's timestamptz type instead of timestamp
2. Documenting this limitation in the PR description

This is particularly problematic since the PR is about adding TIMESTAMP_TZ support, but the timezone information gets silently dropped when writing to PostgreSQL.

            case TIMESTAMP:
            case TIMESTAMP_TZ:
                Integer timestampScale = column.getScale();
                if (timestampScale != null && timestampScale > MAX_TIMESTAMP_SCALE) {
                    timestampScale = MAX_TIMESTAMP_SCALE;
                    log.warn(
                            "The timestamp column {} type timestamp({}) is out of range, "
                                    + "which exceeds the maximum scale of {}, "
                                    + "it will be converted to timestamp({})",
                            column.getName(),
                            column.getScale(),
                            MAX_TIMESTAMP_SCALE,
                            timestampScale);
                }
                if (timestampScale != null && timestampScale > 0) {
                    builder.columnType(String.format("%s(%s)", PG_TIMESTAMP, timestampScale));
                } else {
                    builder.columnType(PG_TIMESTAMP);
                }
                builder.dataType(PG_TIMESTAMP);
                builder.scale(timestampScale);

---

💡 Add Copilot custom instructions for smarter, more guided reviews. Learn how to get started.