Provide fromMicroOfDay convenience init method

Signed-off-by: yuxiqian <[email protected]>

Author: yuxiqian

flink-cdc-common/src/main/java/org/apache/flink/cdc/common/data/TimeData.java

@@ -27,6 +27,7 @@
 public class TimeData implements Comparable<TimeData> {
 
     private static final int SECONDS_TO_MILLIS = 1000;
+    private static final int MILLIS_TO_MICRO = 1000;
     private static final int MILLIS_TO_NANO = 1_000_000;
 
     private final int millisOfDay;
@@ -43,6 +44,10 @@ public static TimeData fromMillisOfDay(int millisOfDay) {
         return new TimeData(millisOfDay);
     }
 
+    public static TimeData fromMicroOfDay(long microOfDay) {
+        return new TimeData((int) (microOfDay / MILLIS_TO_MICRO));
+    }
+
     public static TimeData fromNanoOfDay(long nanoOfDay) {
         // millisOfDay should not exceed 86400000, which is safe to fit into INT.
         return new TimeData((int) (nanoOfDay / MILLIS_TO_NANO));

flink-cdc-connect/flink-cdc-source-connectors/flink-connector-debezium/src/main/java/org/apache/flink/cdc/debezium/event/DebeziumEventDeserializationSchema.java

@@ -320,7 +320,7 @@ protected Object convertToTime(Object dbzObj, Schema schema) {
         if (dbzObj instanceof Long) {
             switch (schema.name()) {
                 case MicroTime.SCHEMA_NAME:
-                    return TimeData.fromNanoOfDay((long) dbzObj * 1000);
+                    return TimeData.fromMicroOfDay((long) dbzObj);
                 case NanoTime.SCHEMA_NAME:
                     return TimeData.fromNanoOfDay((long) dbzObj);
             }