We have an MS SQL connector which has 4 tables whitelisted.
The CDC is enabled on all the tables and we could see the events getting captured in the respective cdc tables.
But these event are not visible in kafka topics. There are no errors in the logs as well.
Note : per logs, we could see that debezium connector is able to identify that CDC is enabled on these tables
**Below is the connector configuration **
curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d'{
"name": "cdc_source_connector",
"config":
{
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"tasks.max": "1",
"database.server.name": "cdc_source_connector",
"table.whitelist": "test.Table1,test.Table2,test.Table3,test.Table4",
"tombstones.on.delete": "false",
"database.history.kafka.topic": "master_data.schema.audit",
"database.history.store.only.monitored.tables.ddl": true,
"include.schema.changes": false,
"event.deserialization.failure.handling.mode": "fail",
"snapshot.mode": "initial_schema_only",
"snapshot.locking.mode": "none",
"snapshot.isolation.mode": "READ_UNCOMMITTED",
"decimal.handling.mode": "string",
"time.precision.mode": "connect",
"transforms": "filtertombstone,addStaticField,topicRoute",
"transforms.filtertombstone.type": "com.***.kafka.connect.transforms.FilterTombstoneRecord",
"transforms.addStaticField.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addStaticField.static.field": "source_system",
"transforms.addStaticField.static.value": "customer",
"transforms.topicRoute.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.topicRoute.regex": "(.*)",
"transforms.topicRoute.replacement": "cdc.all_user_demographics_master_data",
"errors.tolerance": "none",
"errors.log.enable": true,
"errors.log.include.messages": true,
"errors.retry.delay.max.ms": 60000,
"errors.retry.timeout": 300000,
"database.history.kafka.bootstrap.servers": "broker:9092",
"database.hostname": "<>",
"database.port": "1433",
"database.user": "user",
"database.password": "password",
"database.dbname": "dbname"
}
}'