diff --git a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/DateFormatInfo.java b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/DateFormatInfo.java
index 5c2694cf59..48f3161635 100644
--- a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/DateFormatInfo.java
+++ b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/DateFormatInfo.java
@@ -50,11 +50,15 @@ public class DateFormatInfo implements BasicFormatInfo<Date> {
     @Nullable
     private final SimpleDateFormat simpleDateFormat;
 
+    @JsonIgnore
+    @Nullable
+    private final ThreadLocal<SimpleDateFormat> threadLocal;
+
     @JsonCreator
     public DateFormatInfo(
             @JsonProperty(FIELD_FORMAT) @Nonnull String format) {
         this.format = format;
-
+        this.threadLocal = new ThreadLocal<>();
         if (!format.equals("SECONDS")
                 && !format.equals("MILLIS")
                 && !format.equals("MICROS")
@@ -80,6 +84,17 @@ public DateTypeInfo getTypeInfo() {
         return DateTypeInfo.INSTANCE;
     }
 
+    private SimpleDateFormat get() {
+        if (simpleDateFormat == null) {
+            throw new IllegalStateException();
+        }
+        SimpleDateFormat local = threadLocal.get();
+        if (local == null) {
+            threadLocal.set((SimpleDateFormat) simpleDateFormat.clone());
+        }
+        return threadLocal.get();
+    }
+
     @Override
     public String serialize(Date date) {
         switch (format) {
@@ -98,11 +113,7 @@ public String serialize(Date date) {
                 return Long.toString(seconds);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                return simpleDateFormat.format(date);
+                return get().format(date);
             }
         }
     }
@@ -126,11 +137,7 @@ public Date deserialize(String text) throws ParseException {
                 return new Date(millis);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                java.util.Date jDate = simpleDateFormat.parse(text.trim());
+                java.util.Date jDate = get().parse(text.trim());
                 return new Date(jDate.getTime());
             }
         }
diff --git a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimeFormatInfo.java b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimeFormatInfo.java
index 65101cb7bd..8a83592ef4 100644
--- a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimeFormatInfo.java
+++ b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimeFormatInfo.java
@@ -50,6 +50,9 @@ public class TimeFormatInfo implements BasicFormatInfo<Time> {
     @JsonIgnore
     @Nullable
     private final SimpleDateFormat simpleDateFormat;
+    @JsonIgnore
+    @Nullable
+    private final ThreadLocal<SimpleDateFormat> threadLocal;
     @JsonProperty("precision")
     private int precision;
 
@@ -59,6 +62,7 @@ public TimeFormatInfo(
             @JsonProperty("precision") int precision) {
         this.format = format;
         this.precision = precision;
+        this.threadLocal = new ThreadLocal<>();
         if (!format.equals("MICROS")
                 && !format.equals("MILLIS")
                 && !format.equals("SECONDS")
@@ -78,6 +82,17 @@ public TimeFormatInfo() {
         this("HH:mm:ss", DEFAULT_PRECISION_FOR_TIMESTAMP);
     }
 
+    private SimpleDateFormat get() {
+        if (simpleDateFormat == null) {
+            throw new IllegalStateException();
+        }
+        SimpleDateFormat local = threadLocal.get();
+        if (local == null) {
+            threadLocal.set((SimpleDateFormat) simpleDateFormat.clone());
+        }
+        return threadLocal.get();
+    }
+
     @Nonnull
     public String getFormat() {
         return format;
@@ -106,11 +121,7 @@ public String serialize(Time time) {
                 return Long.toString(seconds);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                return simpleDateFormat.format(time);
+                return get().format(time);
             }
         }
     }
@@ -133,11 +144,7 @@ public Time deserialize(String text) throws ParseException {
                 return new Time(millis);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                Date date = simpleDateFormat.parse(text);
+                Date date = get().parse(text);
                 return new Time(date.getTime());
             }
         }
diff --git a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimestampFormatInfo.java b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimestampFormatInfo.java
index 1985c2fea5..c366039f28 100644
--- a/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimestampFormatInfo.java
+++ b/inlong-common/src/main/java/org/apache/inlong/common/pojo/sort/dataflow/field/format/TimestampFormatInfo.java
@@ -52,6 +52,10 @@ public class TimestampFormatInfo implements BasicFormatInfo<Timestamp> {
     @JsonIgnore
     @Nullable
     private final SimpleDateFormat simpleDateFormat;
+    
+    @JsonIgnore
+    @Nullable
+    private final ThreadLocal<SimpleDateFormat> threadLocal;
 
     @JsonProperty("precision")
     private int precision;
@@ -62,6 +66,7 @@ public TimestampFormatInfo(
             @JsonProperty("precision") int precision) {
         this.format = format;
         this.precision = precision;
+        this.threadLocal = new ThreadLocal<>();
         if (!format.equals("MICROS")
                 && !format.equals("MILLIS")
                 && !format.equals("SECONDS")
@@ -85,6 +90,17 @@ public TimestampFormatInfo(@JsonProperty("precision") int precision) {
         this("yyyy-MM-dd HH:mm:ss", precision);
     }
 
+    private SimpleDateFormat get() {
+        if (simpleDateFormat == null) {
+            throw new IllegalStateException();
+        }
+        SimpleDateFormat local = threadLocal.get();
+        if (local == null) {
+            threadLocal.set((SimpleDateFormat) simpleDateFormat.clone());
+        }
+        return threadLocal.get();
+    }
+
     @Nonnull
     public String getFormat() {
         return format;
@@ -114,11 +130,7 @@ public String serialize(Timestamp timestamp) {
                 return Long.toString(seconds);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                return simpleDateFormat.format(timestamp);
+                return get().format(timestamp);
             }
         }
     }
@@ -141,11 +153,7 @@ public Timestamp deserialize(String text) throws ParseException {
                 return new Timestamp(millis);
             }
             default: {
-                if (simpleDateFormat == null) {
-                    throw new IllegalStateException();
-                }
-
-                Date date = simpleDateFormat.parse(text);
+                Date date = get().parse(text);
                 return new Timestamp(date.getTime());
             }
         }