[pipeline-connector][doris][starrocks] Remove SimpleDateFormat for thread safe (#2954)

Author: lvyanquan

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisEventSerializer.java

@@ -35,7 +35,6 @@
 
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.text.SimpleDateFormat;
 import java.time.ZoneId;
 import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
@@ -50,7 +49,8 @@ public class DorisEventSerializer implements DorisRecordSerializer<Event> {
     private Map<TableId, Schema> schemaMaps = new HashMap<>();
 
     /** Format DATE type data. */
-    public static final SimpleDateFormat DATE_FORMATTER = new SimpleDateFormat("yyyy-MM-dd");
+    public static final DateTimeFormatter DATE_FORMATTER =
+            DateTimeFormatter.ofPattern("yyyy-MM-dd");
 
     /** Format timestamp-related type data. */
     public static final DateTimeFormatter DATE_TIME_FORMATTER =

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-doris/src/main/java/com/ververica/cdc/connectors/doris/sink/DorisRowConverter.java

@@ -32,7 +32,6 @@
 
 import java.io.IOException;
 import java.io.Serializable;
-import java.sql.Date;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -100,8 +99,7 @@ static SerializationConverter createExternalConverter(DataType type, ZoneId pipe
                 return (index, val) -> val.getDouble(index);
             case DATE:
                 return (index, val) ->
-                        DATE_FORMATTER.format(
-                                Date.valueOf(LocalDate.ofEpochDay(val.getInt(index))));
+                        LocalDate.ofEpochDay(val.getInt(index)).format(DATE_FORMATTER);
             case TIMESTAMP_WITHOUT_TIME_ZONE:
                 return (index, val) ->
                         val.getTimestamp(index, DataTypeChecks.getPrecision(type))

flink-cdc-connect/flink-cdc-pipeline-connectors/flink-cdc-pipeline-connector-starrocks/src/main/java/com/ververica/cdc/connectors/starrocks/sink/StarRocksUtils.java

@@ -38,8 +38,6 @@
 import com.ververica.cdc.common.types.TinyIntType;
 import com.ververica.cdc.common.types.VarCharType;
 
-import java.sql.Date;
-import java.text.SimpleDateFormat;
 import java.time.LocalDate;
 import java.time.ZoneId;
 import java.time.ZonedDateTime;
@@ -112,7 +110,8 @@ public static void toStarRocksDataType(Column cdcColumn, StarRocksColumn.Builder
     }
 
     /** Format DATE type data. */
-    private static final SimpleDateFormat DATE_FORMATTER = new SimpleDateFormat("yyyy-MM-dd");
+    private static final DateTimeFormatter DATE_FORMATTER =
+            DateTimeFormatter.ofPattern("yyyy-MM-dd");
 
     /** Format timestamp-related type data. */
     private static final DateTimeFormatter DATETIME_FORMATTER =
@@ -168,9 +167,8 @@ record ->
             case DATE:
                 fieldGetter =
                         record ->
-                                DATE_FORMATTER.format(
-                                        Date.valueOf(
-                                                LocalDate.ofEpochDay(record.getInt(fieldPos))));
+                                LocalDate.ofEpochDay(record.getInt(fieldPos))
+                                        .format(DATE_FORMATTER);
                 break;
             case TIMESTAMP_WITHOUT_TIME_ZONE:
                 fieldGetter =