fix: timestamp incompatibility with OceanBase 4.x

docs/content/connectors/oceanbase-cdc(ZH).md

@@ -240,7 +240,7 @@ public class OceanBaseSourceExample {
                       .port(2881)
                       .logProxyHost("127.0.0.1")
                       .logProxyPort(2983)
-                      .serverTimeZone(serverTimezone)
+                      .serverTimeZone(serverTimeZone)
                       .deserializer(deserializer)
                       .build();
 

docs/content/connectors/oceanbase-cdc.md

@@ -406,7 +406,7 @@ public class OceanBaseSourceExample {
                       .port(2881)
                       .logProxyHost("127.0.0.1")
                       .logProxyPort(2983)
-                      .serverTimeZone(serverTimezone)
+                      .serverTimeZone(serverTimeZone)
                       .deserializer(deserializer)
                       .build();
 

flink-connector-oceanbase-cdc/src/main/java/com/ververica/cdc/connectors/oceanbase/source/RowDataOceanBaseDeserializationSchema.java

@@ -43,6 +43,8 @@
 import java.sql.Date;
 import java.sql.Time;
 import java.sql.Timestamp;
+import java.time.Instant;
+import java.time.LocalDateTime;
 import java.time.ZoneId;
 import java.util.Arrays;
 import java.util.Map;
@@ -484,7 +486,10 @@ private static OceanBaseDeserializationRuntimeConverter convertToTimestamp() {
             @Override
             public Object convertSnapshotEvent(Object object) {
                 if (object instanceof Timestamp) {
-                    return TimestampData.fromLocalDateTime(((Timestamp) object).toLocalDateTime());
+                    return TimestampData.fromTimestamp((Timestamp) object);
+                }
+                if (object instanceof LocalDateTime) {
+                    return TimestampData.fromLocalDateTime((LocalDateTime) object);
                 }
                 throw new IllegalArgumentException(
                         "Unable to convert to TimestampData from unexpected value '"
@@ -495,7 +500,11 @@ public Object convertSnapshotEvent(Object object) {
 
             @Override
             public Object convertChangeEvent(String string) {
-                return TimestampData.fromLocalDateTime(Timestamp.valueOf(string).toLocalDateTime());
+                try {
+                    return TimestampData.fromTimestamp(Timestamp.valueOf(string));
+                } catch (IllegalArgumentException e) {
+                    return TimestampData.fromInstant(Instant.parse(string));
+                }
             }
         };
     }
@@ -515,6 +524,10 @@ public Object convertSnapshotEvent(Object object) {
                                     .atZone(serverTimeZone)
                                     .toInstant());
                 }
+                if (object instanceof LocalDateTime) {
+                    return TimestampData.fromInstant(
+                            ((LocalDateTime) object).atZone(serverTimeZone).toInstant());
+                }
                 throw new IllegalArgumentException(
                         "Unable to convert to TimestampData from unexpected value '"
                                 + object
@@ -524,11 +537,15 @@ public Object convertSnapshotEvent(Object object) {
 
             @Override
             public Object convertChangeEvent(String string) {
-                return TimestampData.fromInstant(
-                        Timestamp.valueOf(string)
-                                .toLocalDateTime()
-                                .atZone(serverTimeZone)
-                                .toInstant());
+                try {
+                    return TimestampData.fromInstant(
+                            Timestamp.valueOf(string)
+                                    .toLocalDateTime()
+                                    .atZone(serverTimeZone)
+                                    .toInstant());
+                } catch (IllegalArgumentException e) {
+                    return TimestampData.fromInstant(Instant.parse(string));
+                }
             }
         };
     }