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-12957-c1ff82acf329803e9d570c8e4633253adeb879b7 in repository https://gitbox.apache.org/repos/asf/seatunnel.git
commit 9c50d224e5948fb2291b69b2ba468fc8069e0e2b Author: Ricky Makhija <[email protected]> AuthorDate: Mon Oct 26 13:22:52 2026 +0000 [Fix][Connector-V2] Fix CDC pre-1970 sub-millisecond timestamp conversion (#12957) --- ...TunnelRowDebeziumDeserializationConverters.java | 9 ++- ...elRowDebeziumDeserializationConvertersTest.java | 68 ++++++++++++++++++++++ 2 files changed, 75 insertions(+), 2 deletions(-) diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConverters.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConverters.java index 3e1acd45d0..b6501e7018 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConverters.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/main/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConverters.java @@ -531,11 +531,16 @@ public class SeaTunnelRowDebeziumDeserializationConverters implements Serializab case Timestamp.SCHEMA_NAME: return toLocalDateTime((Long) dbzObj, 0); case MicroTimestamp.SCHEMA_NAME: + // floorMod keeps sub-millis >= 0 for pre-1970 (negative) values long micro = (long) dbzObj; - return toLocalDateTime(micro / 1000, (int) (micro % 1000 * 1000)); + return toLocalDateTime( + Math.floorDiv(micro, 1000L), + (int) (Math.floorMod(micro, 1000L) * 1000)); case NanoTimestamp.SCHEMA_NAME: long nano = (long) dbzObj; - return toLocalDateTime(nano / 1000_000, (int) (nano % 1000_000)); + return toLocalDateTime( + Math.floorDiv(nano, 1000_000L), + (int) Math.floorMod(nano, 1000_000L)); default: } } diff --git a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConvertersTest.java b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConvertersTest.java index e243b8fe17..8051903f54 100644 --- a/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConvertersTest.java +++ b/seatunnel-connectors-v2/connector-cdc/connector-cdc-base/src/test/java/org/apache/seatunnel/connectors/cdc/debezium/row/SeaTunnelRowDebeziumDeserializationConvertersTest.java @@ -19,6 +19,7 @@ package org.apache.seatunnel.connectors.cdc.debezium.row; 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.SeaTunnelDataType; import org.apache.seatunnel.api.table.type.SeaTunnelRow; import org.apache.seatunnel.api.table.type.SeaTunnelRowType; @@ -36,11 +37,16 @@ import org.junit.jupiter.api.Test; import io.debezium.data.geometry.Geography; import io.debezium.data.geometry.Geometry; +import io.debezium.time.MicroTimestamp; +import io.debezium.time.NanoTimestamp; +import io.debezium.time.Timestamp; import java.time.Instant; +import java.time.LocalDateTime; import java.time.OffsetDateTime; import java.time.ZoneId; import java.time.ZoneOffset; +import java.time.temporal.ChronoUnit; import java.util.ArrayList; import java.util.Arrays; import java.util.HashMap; @@ -85,6 +91,68 @@ public class SeaTunnelRowDebeziumDeserializationConvertersTest { Assertions.assertNull(row.getField(1)); } + @Test + void testTimestampConverterBeforeEpoch() throws Exception { + // MySQL DATETIME(4-6) arrives as MicroTimestamp; pre-1970 values are negative epoch micros + SeaTunnelRowDebeziumDeserializationConverters converters = + new SeaTunnelRowDebeziumDeserializationConverters( + new SeaTunnelRowType( + new String[] {"millis", "micros", "nanos"}, + new SeaTunnelDataType[] { + LocalTimeType.LOCAL_DATE_TIME_TYPE, + LocalTimeType.LOCAL_DATE_TIME_TYPE, + LocalTimeType.LOCAL_DATE_TIME_TYPE + }), + new MetadataConverter[] {}, + ZoneId.systemDefault(), + DebeziumDeserializationConverterFactory.DEFAULT); + Schema schema = + SchemaBuilder.struct() + .field("millis", Timestamp.builder().optional().build()) + .field("micros", MicroTimestamp.builder().optional().build()) + .field("nanos", NanoTimestamp.builder().optional().build()) + .build(); + LocalDateTime[] values = { + LocalDateTime.of(2026, 4, 27, 20, 50, 29, 6_733_123), + LocalDateTime.of(1970, 1, 1, 0, 0), + LocalDateTime.of(1969, 12, 31, 23, 59, 59, 999_999_999), + LocalDateTime.of(1969, 12, 31, 12, 0, 0, 500_000_000), + LocalDateTime.of(1900, 6, 15, 12, 30, 45, 123_456_789), + LocalDateTime.of(1000, 1, 1, 0, 0) + }; + for (LocalDateTime value : values) { + Instant instant = value.toInstant(ZoneOffset.UTC); + // Debezium floors to the unit, so encode the same way (ChronoUnit.between truncates) + long seconds = instant.getEpochSecond(); + int nano = instant.getNano(); + // epoch nanos in a long only cover the years 1677 to 2262 + boolean nanosFit = value.getYear() > 1677 && value.getYear() < 2262; + Struct struct = new Struct(schema); + struct.put("millis", seconds * 1000L + nano / 1_000_000); + struct.put("micros", seconds * 1_000_000L + nano / 1_000); + struct.put("nanos", nanosFit ? seconds * 1_000_000_000L + nano : null); + SourceRecord record = + new SourceRecord( + new HashMap<>(), + new HashMap<>(), + "topicName", + null, + SchemaBuilder.int32().build(), + 1, + schema, + struct, + null, + new ArrayList<>()); + + SeaTunnelRow row = converters.convert(record, struct, schema); + Assertions.assertEquals( + value.truncatedTo(ChronoUnit.MILLIS), row.getField(0), "millis"); + Assertions.assertEquals( + value.truncatedTo(ChronoUnit.MICROS), row.getField(1), "micros"); + Assertions.assertEquals(nanosFit ? value : null, row.getField(2), "nanos"); + } + } + @Test void testArrayConverter() throws Exception { DebeziumDeserializationConverter converter;
