(paimon) branch release-0.9 updated: [cdc] Fix mysql cdc '2001-01-01T00:00:00+08:00' parse throw java.time.format.DateTimeParseException (#4428)

[lzljs3620320]

This is an automated email from the ASF dual-hosted git repository.

lzljs3620320 pushed a commit to branch release-0.9
in repository https://gitbox.apache.org/repos/asf/paimon.git


The following commit(s) were added to refs/heads/release-0.9 by this push:
     new 605c3a825 [cdc] Fix mysql cdc '2001-01-01T00:00:00+08:00'  parse throw 
java.time.format.DateTimeParseException (#4428)
605c3a825 is described below

commit 605c3a825d0bc4694feef3687bbcfd778a786e9f
Author: 木鱼和尚 <91652711+lysgithub0...@users.noreply.github.com>
AuthorDate: Mon Nov 4 13:26:46 2024 +0800

    [cdc] Fix mysql cdc '2001-01-01T00:00:00+08:00'  parse throw 
java.time.format.DateTimeParseException (#4428)
---
 .../flink/action/cdc/format/debezium/DebeziumSchemaUtils.java      | 7 ++++++-
 1 file changed, 6 insertions(+), 1 deletion(-)

diff --git 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
index b705bc9e1..564cc8671 100644
--- 
a/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
+++ 
b/paimon-flink/paimon-flink-cdc/src/main/java/org/apache/paimon/flink/action/cdc/format/debezium/DebeziumSchemaUtils.java
@@ -47,6 +47,7 @@ import java.math.BigDecimal;
 import java.nio.ByteBuffer;
 import java.time.Instant;
 import java.time.LocalDateTime;
+import java.time.OffsetDateTime;
 import java.time.ZoneId;
 import java.time.ZoneOffset;
 import java.util.Base64;
@@ -210,7 +211,11 @@ public class DebeziumSchemaUtils {
             // RowDataDebeziumDeserializeSchema#convertToTimestamp in 
flink-cdc-connector
             // for implementation
             LocalDateTime localDateTime =
-                    
Instant.parse(rawValue).atZone(serverTimeZone).toLocalDateTime();
+                    OffsetDateTime.parse(rawValue)
+                            .toInstant()
+                            .atZone(serverTimeZone)
+                            .toLocalDateTime();
+
             transformed = DateTimeUtils.formatLocalDateTime(localDateTime, 6);
         } else if (MicroTime.SCHEMA_NAME.equals(className)) {
             long microseconds = Long.parseLong(rawValue);