kafka connect sql server incremental changes from cdc

Question

I am very new to Kafka (started reading and setting up in my Sandbox environment from just a week) and trying to setup SQL Server JDBC connector. I have setup Confluent community as per confluent guide and installed io.debezium.connector.sqlserver.SqlServerConnector using confluent-hub I enabled CDC on SQL Server Database and required table and it is working fine.

I have tried following connectors (one at-a-time):

io.debezium.connector.sqlserver.SqlServerConnector
io.confluent.connect.jdbc.JdbcSourceConnector

both are loading fine with status of connector and task running fine with no errors as seen below:

Here is my io.confluent.connect.jdbc.JdbcSourceConnector confuguration

{
"name": "mssql-connector",
"config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "mode": "timestamp",
        "timestamp.column.name": "CreatedDateTime",
        "query": "select * from dbo.sampletable",
        "tasks.max": "1",
        "table.types": "TABLE",
        "key.converter.schemas.enable": "false",
        "topic.prefix": "data_",
        "value.converter.schemas.enable": "false",
        "connection.url": "jdbc:sqlserver://SQL2016:1433;databaseName=sampledb",
        "connection.user": "kafka",
        "connection.password": "kafkaPassword@789",
        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
        "key.converter": "org.apache.kafka.connect.json.JsonConverter",
        "poll.interval.ms": "5000",
        "table.poll.interval.ms": "120000"
    }
}

Here is my io.confluent.connect.jdbc.JdbcSourceConnector connector

{
 "name": "mssql-connector",
 "config": {
     "connector.class" : "io.debezium.connector.sqlserver.SqlServerConnector",
     "tasks.max" : "1",
     "database.server.name" : "SQL2016",
     "database.hostname" : "SQL2016",
     "database.port" : "1433",
     "database.user" : "kafka",
     "database.password" : "kafkaPassword@789",
     "database.dbname" : "sampleDb",
     "database.history.kafka.bootstrap.servers" : "kafkanode1:9092",
     "database.history.kafka.topic": "schema-changes.sampleDb"
     }
 }

Both connectors are creating snapshot of a table in a topic (means it pulls all the rows initially) but when I make changes to a table "sampletable" (insert/update/delete), those changes are not being pulled to kafka.

can someone please help me understand how to make CDC working with Kafka?

Thanks

What SQL Server version are you running? Also, can you share example payloads for inserts, deletes, etc. ? — Giorgos Myrianthous
It is SQL Server 2016 Enterprise. I will create sample payload and post it here — Andy Johnson
It would also be useful to provide the steps you followed in order to enable mssql cdc feature. — Giorgos Myrianthous

Andy Johnson Andy Johnson · Accepted Answer · 2020-06-16T14:03:51

This seem to have worked 100%. I am posting answer just in case someone like me stuck on jdbc source connector.

{
"name": "piilog-connector",
"config": {
        "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
        "mode": "incrementing",       
        "value.converter.schemas.enable": "false",
        "connection.url": "jdbc:sqlserver://SQL2016:1433;databaseName=SIAudit",
        "connection.user": "kafka",
        "connection.password": "kafkaPassword@789",  
        "query": "select * from dbo.sampletable",
        "incrementing.column.name": "Id",
        "validate.non.null": false,
        "topic.prefix": "data_",
        "tasks.max": "1",
        "poll.interval.ms": "5000",
        "table.poll.interval.ms": "5000"
    }
}

kafka connect sql server incremental changes from cdc

1 Answers