I'm using the kafka-connect-elasticsearch plugin to get message from my kafka to Elasticsearch. My data in kafka contain a date field (timestamp format).
My first issue was that when I use this plugin, Elasticsearch index didn't recognize the date field as a date type but as a long ... I kind of solve this using an SMT transformation in my connector configuration.
Here is my current configuration that allow me to push data in Elastic :
{
"name": "elasticsearch-sink-test",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "1",
"topics": "test.test",
"key.ignore": "true",
"connection.url": "http://localhost:9200",
"type.name": "kafka-connect",
"name": "elasticsearch-sink-test",
"Batch.size": 100,
"max.buffered.records": 1000,
"Max.retries": 10,
"Retry.backoff.ms": 1000,
"flush.timeout.ms": 20000,
"max.in.flight.requests": 3
"transforms": "date",
"transforms.date.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
"transforms.date.target.type": "Date",
"transforms.date.field": "date",
"transforms.date.format": "yyyy-MM-dd HH:mm:ss"
}
}
My issue right now is that : Elasticsearch doesn't get all the previous messages stored in kafka but only the new one (all the new message that get push to kafka after I started the Elasticsearch connector).
How can I configure the connector to make elastic get all the messages ? Is there any workaround the make elastic "understand" that the date field is a timestamp ?
(For information, my data origin are a MongoDB with the debezium CDC connector)
Thank you in advance