mysql-sink.json

{
    "name": "test-mysql-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "group.id": "test-mysql-sink",
        "topics": "post,comment",
        "connection.url": "jdbc:mysql://localhost:3306/test-des?user=root&nullCatalogMeansCurrent=true",
        "auto.create": "true",
        "auto.evolve": "true",
        "insert.mode": "upsert",
        "pk.fields": "id",
        "pk.mode": "record_key",
        "delete.enabled": "true",
        "value.converter": "io.confluent.connect.avro.AvroConverter",
        "value.converter.schema.registry.url": "http://localhost:8081",
        "key.converter": "io.confluent.connect.avro.AvroConverter",
        "key.converter.schema.registry.url": "http://localhost:8081",
        "dialect.name": "MySqlDatabaseDialect",
        "transforms": "unwrap,created_at,updated_at",
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
        "transforms.unwrap.drop.tombstones": "false ",
        "transforms.created_at.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.created_at.target.type": "Timestamp",
        "transforms.created_at.field": "created_at",
        "transforms.created_at.format": "yyyy-MM-dd'T'HH:mm:ss'Z'",
        "transforms.updated_at.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
        "transforms.updated_at.target.type": "Timestamp",
        "transforms.updated_at.field": "updated_at",
        "transforms.updated_at.format": "yyyy-MM-dd'T'HH:mm:ss'Z'"
      }
}