I am using Kafka to replicate a mysql database from one instance to another using debezium CDC, below is my sink connector configuration, where i use "auto.create:true" which actually creates the destination tables automatically BUT the data-type between source and sink DBs do not match. For example in source DB, a table column has data type as timestamp but auto.create feature creates it as varchar(256) type instead of datetime.
source value is 2020-02-10 13:34:30 (datetime type) and sink table value is created as 2020-02-10T08:34:30Z (varchar(256) type)
Similar is the case for date and datetime where columns of int types are created.
1) How can I configure it to have exact data types as source DB??
2) Also if I create the sink tables manually (auto.create:false) with exact data-types as in source db (i.e timestamp), upon snapshot/insertion in sink DB, I get an error some mysql Data truncation: Incorrect datetime value: '2020-05-09T11:10:01Z' as column type is datetime but the data connector is sending is 2020-02-10T08:34:30Z, is there any way that the sink connector converts this string data as per correct datetime value (have multiple tables and columns to follow same flow)??
Sink Connector config:
{
"name": "sink-connector",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"tasks.max": "1",
"connection.url": "jdbc:mysql://host:3306/db?nullCatalogMeansCurrent=true",
"connection.user": "",
"connection.password": "",
"topics": "topic1,topic2,topic3....",
"transforms": "unwrap",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.delete.handling.mode": "rewrite",
"insert.mode": "insert",
"auto.create": "true",
"auto.evolve": "true",
"delete.enabled": "false",
"pk.mode": "none"
}
}