kafka jdbc sink connector creates invalid data type

Viewed 373

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"
    }
}
0 Answers
Related