[Fix][Connector-V2] Fixed missing timestamp accuracy of starrocks connector (#9096)

Author: corgy-w

seatunnel-connectors-v2/connector-starrocks/src/main/java/org/apache/seatunnel/connectors/seatunnel/starrocks/serialize/StarRocksBaseSerializer.java

@@ -19,25 +19,29 @@
 
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.common.exception.CommonErrorCodeDeprecated;
-import org.apache.seatunnel.common.utils.DateTimeUtils;
 import org.apache.seatunnel.common.utils.DateUtils;
 import org.apache.seatunnel.common.utils.JsonUtils;
 import org.apache.seatunnel.common.utils.TimeUtils;
 import org.apache.seatunnel.connectors.seatunnel.starrocks.exception.StarRocksConnectorException;
 
-import lombok.Builder;
-
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.LocalTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
 
 public class StarRocksBaseSerializer {
-    @Builder.Default private DateUtils.Formatter dateFormatter = DateUtils.Formatter.YYYY_MM_DD;
+    private DateUtils.Formatter dateFormatter = DateUtils.Formatter.YYYY_MM_DD;
 
-    @Builder.Default
-    private DateTimeUtils.Formatter dateTimeFormatter = DateTimeUtils.Formatter.YYYY_MM_DD_HH_MM_SS;
+    private DateTimeFormatter dateTimeFormatter =
+            new DateTimeFormatterBuilder()
+                    .appendPattern("yyyy-MM-dd HH:mm:ss")
+                    .optionalStart()
+                    .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+                    .toFormatter();
 
-    @Builder.Default private TimeUtils.Formatter timeFormatter = TimeUtils.Formatter.HH_MM_SS;
+    private TimeUtils.Formatter timeFormatter = TimeUtils.Formatter.HH_MM_SS;
 
     protected Object convert(SeaTunnelDataType dataType, Object val) {
         if (val == null) {
@@ -59,7 +63,7 @@ protected Object convert(SeaTunnelDataType dataType, Object val) {
             case TIME:
                 return TimeUtils.toString((LocalTime) val, timeFormatter);
             case TIMESTAMP:
-                return DateTimeUtils.toString((LocalDateTime) val, dateTimeFormatter);
+                return ((LocalDateTime) val).format(dateTimeFormatter);
             case ARRAY:
             case MAP:
                 return JsonUtils.toJsonString(val);

seatunnel-connectors-v2/connector-starrocks/src/test/java/org/apache/seatunnel/connectors/seatunnel/starrocks/serialize/StarRocksJsonSerializerTest.java

@@ -19,6 +19,7 @@
 
 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.MapType;
 import org.apache.seatunnel.api.table.type.SeaTunnelDataType;
 import org.apache.seatunnel.api.table.type.SeaTunnelRow;
@@ -27,30 +28,46 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
+import java.time.format.DateTimeFormatterBuilder;
+import java.time.temporal.ChronoField;
 import java.util.Collections;
 
 public class StarRocksJsonSerializerTest {
 
+    private DateTimeFormatter dateTimeFormatter =
+            new DateTimeFormatterBuilder()
+                    .appendPattern("yyyy-MM-dd HH:mm:ss")
+                    .optionalStart()
+                    .appendFraction(ChronoField.NANO_OF_SECOND, 0, 9, true)
+                    .toFormatter();
+
     @Test
     public void serialize() {
-        String[] filedNames = {"id", "name", "array", "map"};
-        SeaTunnelDataType<?>[] filedTypes = {
+        String[] fieldNames = {"id", "name", "array", "map", "timestamp"};
+        SeaTunnelDataType<?>[] fieldTypes = {
             BasicType.LONG_TYPE,
             BasicType.STRING_TYPE,
             ArrayType.STRING_ARRAY_TYPE,
-            new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE)
+            new MapType<>(BasicType.STRING_TYPE, BasicType.STRING_TYPE),
+            LocalTimeType.LOCAL_DATE_TIME_TYPE
         };
 
-        SeaTunnelRowType seaTunnelRowType = new SeaTunnelRowType(filedNames, filedTypes);
+        SeaTunnelRowType seaTunnelRowType = new SeaTunnelRowType(fieldNames, fieldTypes);
         StarRocksJsonSerializer starRocksJsonSerializer =
                 new StarRocksJsonSerializer(seaTunnelRowType, false);
         Object[] fields = {
-            1, "Tom", new String[] {"tag1", "tag2"}, Collections.singletonMap("key1", "value1")
+            1,
+            "Tom",
+            new String[] {"tag1", "tag2"},
+            Collections.singletonMap("key1", "value1"),
+            LocalDateTime.parse("2024-01-25 07:55:45.123", dateTimeFormatter)
         };
         SeaTunnelRow seaTunnelRow = new SeaTunnelRow(fields);
         String jsonString = starRocksJsonSerializer.serialize(seaTunnelRow);
         Assertions.assertEquals(
-                "{\"id\":1,\"name\":\"Tom\",\"array\":[\"tag1\",\"tag2\"],\"map\":{\"key1\":\"value1\"}}",
+                "{\"id\":1,\"name\":\"Tom\",\"array\":[\"tag1\",\"tag2\"],\"map\":{\"key1\":\"value1\"},\"timestamp\":\"2024-01-25 07:55:45.123\"}",
                 jsonString);
     }
 }