# Debezium MSSQL CDC with Kafka (KRaft)
# Debezium MySQL CDC with Kafka (KRaft)
# 1. Download Correct Debezium SQL Server Plugin (v3.3)
```
wget https://repo1.maven.org/maven2/io/debezium/debezium-connector-sqlserver/3.3.0.Final/debezium-connector-sqlserver-3.3.0.Final-plugin.tar.gz
```
# Extract plugin And Move to /opt/kafka/connectors
```
tar -xzf debezium-connector-sqlserver-3.3.0.Final-plugin.tar.gz -C /opt/kafka/connectors
```
# Restart your Kafka Connect service:
`systemctl restart kafka-connect`
# Create SQL Server User With Required Permissions
```sql
CREATE DATABASE DebeziumTest;
USE DebeziumTest;
EXEC sys.sp_cdc_enable_db;
CREATE TABLE dbo.Customers (
CustomerID INT IDENTITY(1,1) PRIMARY KEY,
FirstName NVARCHAR(100),
LastName NVARCHAR(100),
Email NVARCHAR(255),
CreatedAt DATETIME DEFAULT GETDATE()
);
EXEC sys.sp_cdc_enable_table
@source_schema = 'dbo',
@source_name = 'Customers',
@role_name = NULL,
@supports_net_changes = 0;
INSERT INTO dbo.Customers (FirstName, LastName, Email)
VALUES
('Golam', 'Kibria', 'golam@example.com'),
('John', 'Doe', 'john.doe@example.com'),
('Jane', 'Smith', 'jane.smith@example.com');
UPDATE dbo.Customers
SET Email = 'golam.kibria@updated.com'
WHERE CustomerID = 1;
SELECT name, is_cdc_enabled FROM sys.databases WHERE name='DebeziumTest';
SELECT name, is_tracked_by_cdc FROM sys.tables WHERE name='Customers';
USE master;
CREATE LOGIN debezium WITH PASSWORD = 'StrongPassword!123';
USE DebeziumTest;
CREATE USER debezium FOR LOGIN debezium;
ALTER ROLE db_owner ADD MEMBER debezium;
```
# Register debezium connector
```bash
curl -X POST http://192.168.1.226:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "mssql-connector",
"config": {
"connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
"database.hostname": "192.168.1.225",
"database.port": "1433",
"database.user": "debezium",
"database.password": "StrongPassword!123",
"database.names": "DebeziumTest",
"database.server.name": "sqlserver",
"database.encrypt": "false",
"topic.prefix": "mssql",
"snapshot.mode": "initial",
"table.include.list": "dbo.Customers",
"schema.history.internal.kafka.bootstrap.servers": "192.168.1.226:9092",
"schema.history.internal.kafka.topic": "schema-changes.mssql",
"include.schema.changes": "true"
}
}'
```
# . Verify Connector
```bash
curl http://localhost:8083/connectors
```
# Check status
```
curl http://localhost:8083/connectors/mssql-connector/status
```
# . Kafka Topics Created
```bash
/opt/kafka/bin/kafka-topics.sh --bootstrap-server localhost:9092 --list
```
# Consume CDC Messages
```
/opt/kafka/bin/kafka-console-consumer.sh \
--bootstrap-server 192.168.1.226:9092 \
--topic mssql.sqlserver.DebeziumTest.dbo.Customers \
--from-beginning
```
# Delete Connector
```
curl -X DELETE http://localhost:8083/connectors/mssql-connector
```
# Check kafka-connect log
```
tail -f /opt/kafka/logs/connect.log
```
# Create User and permissions
# ✅ SQL Server Debezium Test — Without `GO`
### **1. Create Database**
### **Switch Database**
---
# ✅ 2. Enable CDC on the Database
(Works without GO)
---
# ✅ 3. Create Test Table
---
# ✅ 4. Enable CDC on the Table
---
# ✅ 5. Insert Sample Data
---
# ✅ 6. Update Data
---
# ✅ 7. Delete Data
---
# 🔍 8. Check CDC Status