[Bug] [Connector-V2][ES] timestamp type conversion fails and sink stage there will be redundant "

[kpretty]

Search before asking

• I had searched in the issues and found no similar issues.

What happened

timestamp for es

SeaTunnel Version

2.3.0

SeaTunnel Config

env {
  execution.parallelism = 8
  job.mode = "BATCH"
  job.name = "opensearch to console"
}

source {
  Elasticsearch {
    hosts = ["https://xxx:9200"]
    index = "arkime_sessions3*"
    source = ["firstPacket","event.id","event.kind","source.ip","client.nat.ip","source.port","client.nat.port","source.mac","source.hostname","destination.ip","server.nat.ip","destination.port","destination.nat.port","destination.mac","destination.hostname","client.user.name","client.user.domain","client.user.id","host.ip","suricata.alert.severity","rule.name","event.action","event.result","http.response.bytes","event.sequence","network.protocol","network.application","suricata.http.status","suricata.http.http_method","suricata.http.url","suricata.http.hostname","suricata.http.http_user_agent","http.request.referrer","user_agent.device.name","http.authTypeCnt","zeek.smb_cmd.command","file.path","threat.technique.name","related.oui","file.pe.product","suricata.flow.start","event.duration","event.type","host.geo.country_name","host.geo.region_name","host.geo.city_name","source.geo.country_name","source.geo.region_name","source.geo.city_name","source.geo.longitude","source.geo.latitude","destination.geo.country_name","destination.geo.region_name","destination.geo.city_name","destination.geo.longitude","destination.geo.latitude","event.original","network.direction","tags","event.hash"]
    username = "xxx"
    password = "xxx"
    tls_keystore_path = "opensearch.keystroe"
    tls_verify_hostname = false
  }  
}

transform {

}

sink {
    Console {}

}

Running Command

bin/seatunnel.sh --config ../seatunneljob/job-es-no-transform-to-console.conf -m local

Error Exception

Caused by: java.time.format.DateTimeParseException: Text '1065823369239' could not be parsed at index 0
	at java.time.format.DateTimeFormatter.parseResolved0(DateTimeFormatter.java:1949)
	at java.time.format.DateTimeFormatter.parse(DateTimeFormatter.java:1851)
	at java.time.LocalDateTime.parse(LocalDateTime.java:492)
	at org.apache.seatunnel.connectors.seatunnel.elasticsearch.serialize.source.DefaultSeaTunnelRowDeserializer.parseDate(DefaultSeaTunnelRowDeserializer.java:209)
	at org.apache.seatunnel.connectors.seatunnel.elasticsearch.serialize.source.DefaultSeaTunnelRowDeserializer.convertValue(DefaultSeaTunnelRowDeserializer.java:161)
	at org.apache.seatunnel.connectors.seatunnel.elasticsearch.serialize.source.DefaultSeaTunnelRowDeserializer.convert(DefaultSeaTunnelRowDeserializer.java:121)
	... 13 more

	at org.apache.seatunnel.engine.client.job.ClientJobProxy.waitForJobComplete(ClientJobProxy.java:118)
	at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:151)
	... 2 more

Flink or Spark Version

No response

Java or Scala Version

No response

Screenshots

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[kpretty]

I have submitted a PR #4233

[github-actions]

This issue has been automatically marked as stale because it has not had recent activity for 30 days. It will be closed in next 7 days if no further activity occurs.