sinkconnector.json

{
    "name": "jdbc-sink",
    "config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
        "tasks.max": "1",
        "topics": "inventory.customers",
        "connection.url": "jdbc:postgresql://postgres:5432/inventory?user=postgresuser&password=postgrespw",
        "transforms": "unwrap,route",           
        "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",  
        "transforms.unwrap.drop.tombstones":"false",
        "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
        "transforms.route.regex": "([^.]+)\\.([^.]+)",
        "transforms.route.replacement": "$2",
        "auto.create": "true",                                                   
        "insert.mode": "upsert",    
        "delete.enabled": "true",
        "pk.fields": "id",   
        "table.whitelist":"customers",
        "pk.mode": "record_key"                                               
    }
}