[FLINK-39209][doris] Fix time data type serialiazation when sink to doris with pipeline connector (apache#4312)

Author: chengcongchina

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializer.java

@@ -67,6 +67,13 @@ public class DorisEventSerializer implements DorisRecordSerializer<Event> {
     public static final DateTimeFormatter DATE_TIME_FORMATTER =
             DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss.SSSSSS");
 
+    /** Format TIME type data without precision. */
+    public static final DateTimeFormatter TIME_FORMATTER = DateTimeFormatter.ofPattern("HH:mm:ss");
+
+    /** Format TIME type data with millisecond precision. */
+    public static final DateTimeFormatter TIME_WITH_MILLISECOND_FORMATTER =
+            DateTimeFormatter.ofPattern("HH:mm:ss.SSS");
+
     /** ZoneId from pipeline config to support timestamp with local time zone. */
     public final ZoneId pipelineZoneId;
 

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/org/apache/flink/cdc/connectors/doris/sink/DorisRowConverter.java

@@ -117,7 +117,18 @@ static SerializationConverter createExternalConverter(DataType type, ZoneId pipe
                 final int zonedP = ((ZonedTimestampType) type).getPrecision();
                 return (index, val) -> val.getTimestamp(index, zonedP).toTimestamp();
             case TIME_WITHOUT_TIME_ZONE:
-                return (index, val) -> val.getTime(index).toLocalTime();
+                return (index, val) -> {
+                    int precision = DataTypeChecks.getPrecision(type);
+                    if (precision == 0) {
+                        return val.getTime(index)
+                                .toLocalTime()
+                                .format(DorisEventSerializer.TIME_FORMATTER);
+                    } else {
+                        return val.getTime(index)
+                                .toLocalTime()
+                                .format(DorisEventSerializer.TIME_WITH_MILLISECOND_FORMATTER);
+                    }
+                };
             case ARRAY:
                 return (index, val) -> convertArrayData(val.getArray(index), type);
             case MAP:

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/test/java/org/apache/flink/cdc/connectors/doris/sink/DorisEventSerializerTest.java

@@ -19,6 +19,7 @@
 
 import org.apache.flink.cdc.common.configuration.Configuration;
 import org.apache.flink.cdc.common.data.DateData;
+import org.apache.flink.cdc.common.data.TimeData;
 import org.apache.flink.cdc.common.data.TimestampData;
 import org.apache.flink.cdc.common.data.binary.BinaryStringData;
 import org.apache.flink.cdc.common.event.CreateTableEvent;
@@ -41,6 +42,7 @@
 import java.time.Instant;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
+import java.time.LocalTime;
 import java.time.ZoneId;
 import java.util.HashMap;
 import java.util.Map;
@@ -67,6 +69,38 @@ public class DorisEventSerializerTest {
     private static final BinaryRecordDataGenerator RECORD_DATA_GENERATOR =
             new BinaryRecordDataGenerator(((RowType) SCHEMA.toRowDataType()));
 
+    @Test
+    public void testDataChangeEventWithTimeDataType() throws IOException {
+        Schema schema =
+                Schema.newBuilder()
+                        .physicalColumn("id_", DataTypes.BIGINT().notNull())
+                        .physicalColumn("time_0_", DataTypes.TIME(0))
+                        .physicalColumn("time_3_", DataTypes.TIME(3))
+                        .primaryKey("id_")
+                        .build();
+        BinaryRecordDataGenerator generator =
+                new BinaryRecordDataGenerator(((RowType) schema.toRowDataType()));
+        CreateTableEvent createTableEvent = new CreateTableEvent(TABLE_ID, schema);
+        DataChangeEvent dataChangeEvent =
+                DataChangeEvent.insertEvent(
+                        TABLE_ID,
+                        generator.generate(
+                                new Object[] {
+                                    1L,
+                                    TimeData.fromLocalTime(LocalTime.of(19, 43, 17)),
+                                    TimeData.fromLocalTime(LocalTime.of(21, 45, 3, 123000000)),
+                                }));
+
+        dorisEventSerializer = new DorisEventSerializer(ZoneId.of("UTC"), new Configuration());
+        dorisEventSerializer.serialize(createTableEvent);
+        DorisRecord dorisRecord = dorisEventSerializer.serialize(dataChangeEvent);
+        JsonNode jsonNode = objectMapper.readTree(dorisRecord.getRow());
+
+        Assertions.assertThat(jsonNode.get("id_").asLong()).isEqualTo(1L);
+        Assertions.assertThat(jsonNode.get("time_0_").asText()).isEqualTo("19:43:17");
+        Assertions.assertThat(jsonNode.get("time_3_").asText()).isEqualTo("21:45:03.123");
+    }
+
     @Test
     public void testDataChangeEventWithDateTimePartitionColumn() throws IOException {
         Map<String, String> configMap = new HashMap<>();