[Bug][Connector-v2][ElasticsearchSource] Fix behavior when source empty.

[Improve][Connector-v2][ElasticsearchSource] Support SourceConfig.SOURCE field empty.
CosmosNi · Mar 26, 2024 · fbb4e26 · fbb4e26
1 parent f2d86c1
commit fbb4e26
Showing 1 changed file with 7 additions and 0 deletions.
diff --git a/...h-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java b/...h-e2e/src/test/java/org/apache/seatunnel/e2e/connector/elasticsearch/ElasticsearchIT.java
@@ -48,10 +48,12 @@
 
 import java.io.IOException;
 import java.math.BigDecimal;
+import java.text.DateFormat;
 import java.time.Duration;
 import java.time.LocalDate;
 import java.time.LocalDateTime;
 import java.time.ZoneOffset;
+import java.time.format.DateTimeFormatter;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.Collections;
@@ -218,6 +220,11 @@ private List<String> readSinkDataWithOutSchema() throws InterruptedException {
                             x.remove("_index");
                             x.remove("_type");
                             x.remove("_id");
+                            x.replace(
+                                    "c_timestamp",
+                                    LocalDateTime.parse(String.valueOf(x.get("c_timestamp")))
+                                            .toInstant(ZoneOffset.UTC)
+                                            .toEpochMilli());
                         });
         List<String> docs =
                 scrollResult.getDocs().stream()