[Improve][Elasticsearch] Add LocalDateTime serialization test and simplify serializer

seatunnel-connectors-v2/connector-elasticsearch/src/main/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializer.java

@@ -36,6 +36,9 @@
 import lombok.NonNull;
 
 import java.nio.ByteBuffer;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.format.DateTimeFormatter;
 import java.time.temporal.Temporal;
 import java.util.Collections;
 import java.util.HashMap;
@@ -45,6 +48,12 @@
 
 /** use in elasticsearch version >= 2.x and <= 8.x */
 public class ElasticsearchRowSerializer implements SeaTunnelRowSerializer {
+    // Use ISO 8601 format which is compatible with Elasticsearch's strict_date_optional_time
+    private static final DateTimeFormatter LOCAL_DATE_TIME_FORMATTER =
+            DateTimeFormatter.ofPattern("yyyy-MM-dd'T'HH:mm:ss");
+    private static final DateTimeFormatter LOCAL_DATE_FORMATTER =
+            DateTimeFormatter.ofPattern("yyyy-MM-dd");
+
     private final SeaTunnelRowType seaTunnelRowType;
     private final ObjectMapper objectMapper = new ObjectMapper();
 
@@ -203,6 +212,12 @@ private Map<String, Object> toDocumentMap(SeaTunnelRow row, SeaTunnelRowType row
     private Object convertValue(String fieldName, Object value) {
         if (value instanceof Temporal) {
             // jackson not support jdk8 new time api
+            if (value instanceof LocalDateTime) {
+                // Use ISO 8601 format compatible with Elasticsearch's strict_date_optional_time
+                return ((LocalDateTime) value).format(LOCAL_DATE_TIME_FORMATTER);
+            } else if (value instanceof LocalDate) {
+                return ((LocalDate) value).format(LOCAL_DATE_FORMATTER);
+            }
             return value.toString();
         } else if (value instanceof Map) {
             for (Map.Entry<?, ?> entry : ((Map<?, ?>) value).entrySet()) {

seatunnel-connectors-v2/connector-elasticsearch/src/test/java/org/apache/seatunnel/connectors/seatunnel/elasticsearch/serialize/ElasticsearchRowSerializerTest.java

@@ -31,6 +31,7 @@
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.api.Test;
 
+import java.time.LocalDateTime;
 import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Map;
@@ -184,4 +185,33 @@ public void testSerializeDelete() {
         String upsertStr = serializer.serializeRow(row);
         Assertions.assertEquals(expected, upsertStr);
     }
+
+    @Test
+    public void testSerializeLocalDateTimeFieldFormat() {
+        String index = "st_index";
+        Map<String, Object> confMap = new HashMap<>();
+        confMap.put(ElasticsearchSinkOptions.INDEX.key(), index);
+
+        ReadonlyConfig pluginConf = ReadonlyConfig.fromMap(confMap);
+        ElasticsearchClusterInfo clusterInfo =
+                ElasticsearchClusterInfo.builder().clusterVersion("8.0.0").build();
+        IndexInfo indexInfo = new IndexInfo(index, pluginConf);
+        SeaTunnelRowType schema =
+                new SeaTunnelRowType(
+                        new String[] {"id", "ts"},
+                        new SeaTunnelDataType[] {STRING_TYPE, STRING_TYPE});
+
+        final ElasticsearchRowSerializer serializer =
+                new ElasticsearchRowSerializer(clusterInfo, indexInfo, schema);
+
+        String id = "0001";
+        LocalDateTime ts = LocalDateTime.of(2023, 1, 2, 3, 4, 5);
+        SeaTunnelRow row = new SeaTunnelRow(new Object[] {id, ts});
+        row.setRowKind(RowKind.UPDATE_AFTER);
+
+        String result = serializer.serializeRow(row);
+        Assertions.assertTrue(
+                result.contains("\"ts\":\"2023-01-02T03:04:05\""),
+                "LocalDateTime field should be formatted with ISO-8601 'T' separator");
+    }
 }