Setting Up Debezium CDC with Kafka on Windows

This article explains how to set up a complete Debezium CDC environment on Windows bare metal using Kafka, Kafka Connect, Schema Registry, and Kafka UI. It covers installation, configuration, service startup, connector creation, and topic configuration using simple step-by-step instructions. By the end of this guide, MySQL source database changes will be captured using Debezium and published automatically to Kafka topics for real-time data streaming and processing.

What is Debezium?

Debezium is an open-source Change Data Capture (CDC) platform used to capture database changes in real time. It monitors databases such as MySQL, PostgreSQL, SQL Server, and Oracle, and detects operations like INSERT, UPDATE, and DELETE. Whenever data changes in the source database, Debezium reads those changes from the database transaction logs and sends them to Kafka topics automatically. This allows applications and downstream systems to react to database changes in real time without directly querying the database continuously.

What is Kafka?

Apache Kafka is a distributed event streaming platform used to store and transfer real-time data streams. Kafka works using topics. Producers publish messages to Kafka topics, and consumers read those messages from the topics. Kafka is highly scalable, fault tolerant, and capable of handling large amounts of real-time data. It is commonly used for event-driven systems, data pipelines, log processing, and CDC solutions. In this setup, Debezium publishes database change events into Kafka topics.

What is Kafka Connect?

Kafka Connect is a framework provided by Kafka to move data between Kafka and external systems. It allows connectors to be configured without writing custom code. Kafka Connect supports both:

• Source Connectors → Read data from external systems into Kafka
• Sink Connectors → Send data from Kafka to external systems

In this setup, the Debezium MySQL connector runs inside Kafka Connect and is responsible for capturing database changes and publishing them to Kafka topics.

What is Schema Registry?

Confluent Schema Registry is a service used to manage and store message schemas. When Kafka messages are serialized using formats such as Avro, the schema information is stored in Schema Registry. This helps producers and consumers validate message structures and maintain compatibility between different versions of schemas. Schema Registry helps prevent issues caused by schema changes and ensures reliable data exchange between systems.

What is Kafka UI?

Kafka UI is a web-based interface used to manage and monitor Kafka clusters. It provides an easy way to:

• View Kafka topics
• Browse messages
• Create and manage Kafka connectors
• Monitor consumer groups
• View schemas from Schema Registry

Kafka UI simplifies Kafka administration and helps visualize the data flowing through the Kafka ecosystem.

Let’s get started ✌️

Prerequisites

• Configure the JAVA_HOME system variable.
• Add the Java bin directory to the system PATH variable.
• Update MySQL Configuration File.

Edit the MySQL configuration file (my.cnf or my.ini) on the source database server and add or update the following configurations to enable the bin-logs. Debezium requires bin-logs to be enabled in the source MySQL database to identify data changes.

server-id=12345
log_bin=mysql-bin
binlog_format=ROW
binlog_row_image=FULL
binlog_expire_logs_seconds=604800

Kafka

Install Kafka

Download the latest Kafka release from the URL below. Download the .tgz file from the list.

https://downloads.apache.org/kafka/Kafka can be extracted to any preferred location. In this guide, C:\kafka will be used.

Create a folder named kafka in the C:\ drive. Then execute the following command to extract the Kafka .tgz file into the C:\kafka directory.

tar -xvzf kafka_2.13-4.2.0.tgz -C C:\kafka --strip-components=1

Configure Kafka

• Create a folder named kraft inside the C:\kafka\config\ directory.
• Create a file named server.properties inside the C:\kafka\config\kraft directory and add the following configurations.

########################
# KRaft MODE
########################
process.roles=broker,controller
node.id=1
controller.quorum.voters=1@localhost:9093
########################
# LISTENERS
########################
listeners=PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093
advertised.listeners=PLAINTEXT://localhost:9092
listener.security.protocol.map=CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
inter.broker.listener.name=PLAINTEXT
controller.listener.names=CONTROLLER
########################
# STORAGE
########################
log.dirs=C:/kafka/data
########################
# SINGLE BROKER
########################
num.partitions=3
default.replication.factor=1
min.insync.replicas=1
offsets.topic.replication.factor=1
transaction.state.log.replication.factor=1
transaction.state.log.min.isr=1
########################
# PERFORMANCE
########################
num.network.threads=4
num.io.threads=8
message.max.bytes=10485760
########################
# RETENTION
########################
log.retention.hours=720
log.segment.bytes=1073741824
log.retention.bytes=107374182400
log.cleanup.policy=compact,delete

• Create a folder named data inside the C:\kafka\ directory.
• Create a folder named plugins inside the C:\kafka\ directory.
• Edit the connect-distributed.properties file located in C:\kafka\config. Uncomment the plugin.path property and set it as shown below.

plugin.path=C:/kafka/plugins

• Download the kafka-connect plugins from the following GitHub repository.

https://github.com/kanishkamherath/kafka-connect-plugins

• Copy the avro-converter and debezium-connector-mysql folders to C:\kafka\plugins directory.

The debezium-connector-mysql plugin can also be downloaded separately from the Debezium repository using the URL below.

https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/3.4.2.Final/debezium-connector-mysql-3.4.2.Final-plugin.tar.gz

• Create a batch file named kafka-init.bat in the C:\kafka directory and add the following content to it.

@echo off
setlocal EnableExtensions EnableDelayedExpansion

REM =============================================================================
REM Kafka KRaft One-Time Initialization Script (Windows .bat)
REM Run this ONCE before starting Kafka (service or manual start)
REM =============================================================================

REM ---------------------------------------------------------------------------
REM CONFIGURATION
REM ---------------------------------------------------------------------------
set "KAFKA_HOME=C:\kafka"
set "KAFKA_CONFIG=%KAFKA_HOME%\config\kraft\server.properties"
set "KAFKA_LOG_DIR=%KAFKA_HOME%\logs"
set "CLUSTER_ID_FILE=%KAFKA_HOME%\config\kraft\.cluster_id"

set "PREFIX_INFO=[INFO] "
set "PREFIX_WARN=[WARN] "
set "PREFIX_ERR=[ERROR] "

echo %PREFIX_INFO%Running pre-checks...

REM ---------------------------------------------------------------------------
REM PRE-CHECKS
REM ---------------------------------------------------------------------------

REM Require Administrator
whoami /groups | findstr /i /c:"S-1-5-32-544" >nul 2>&1
if NOT %ERRORLEVEL%==0 (
  echo %PREFIX_ERR%Please run this script as Administrator.
  exit /b 1
)

REM Check Kafka is installed
if not exist "%KAFKA_HOME%\bin\windows\kafka-storage.bat" (
  echo %PREFIX_ERR%Kafka not found at "%KAFKA_HOME%". Update KAFKA_HOME variable.
  exit /b 1
)

REM Check config file exists
if not exist "%KAFKA_CONFIG%" (
  echo %PREFIX_ERR%server.properties not found at "%KAFKA_CONFIG%".
  exit /b 1
)

REM Check if already formatted - prevent accidental data wipe
if exist "%CLUSTER_ID_FILE%" (
  echo %PREFIX_WARN%Kafka storage appears to already be initialized.
  echo %PREFIX_WARN%Cluster ID file found at: "%CLUSTER_ID_FILE%"
  echo %PREFIX_WARN%Skipping format step to avoid data loss.
  echo %PREFIX_WARN%If you intended to reinitialize, delete "%CLUSTER_ID_FILE%" and "%KAFKA_LOG_DIR%" manually first.
  exit /b 0
)

REM ---------------------------------------------------------------------------
REM STEP 1 - Ensure log directory exists
REM ---------------------------------------------------------------------------
echo %PREFIX_INFO%Step 1/3 - Preparing log directory: "%KAFKA_LOG_DIR%"

if not exist "%KAFKA_LOG_DIR%" (
  mkdir "%KAFKA_LOG_DIR%"
  if NOT %ERRORLEVEL%==0 (
    echo %PREFIX_ERR%Failed to create directory: "%KAFKA_LOG_DIR%"
    exit /b 1
  )
  echo %PREFIX_INFO%Created directory: "%KAFKA_LOG_DIR%"
) else (
  echo %PREFIX_INFO%Directory already exists: "%KAFKA_LOG_DIR%"
)

REM ---------------------------------------------------------------------------
REM STEP 2 - Generate Cluster ID
REM ---------------------------------------------------------------------------
echo %PREFIX_INFO%Step 2/3 - Generating Kafka Cluster ID...

set "KAFKA_CLUSTER_ID="
for /f "usebackq delims=" %%I in (`"%KAFKA_HOME%\bin\windows\kafka-storage.bat" random-uuid`) do set "KAFKA_CLUSTER_ID=%%I"

if "%KAFKA_CLUSTER_ID%"=="" (
  echo %PREFIX_ERR%Failed to generate Cluster ID.
  exit /b 1
)

echo %PREFIX_INFO%Cluster ID generated: %KAFKA_CLUSTER_ID%

REM Save cluster ID to file for future reference
> "%CLUSTER_ID_FILE%" (echo %KAFKA_CLUSTER_ID%)
if NOT %ERRORLEVEL%==0 (
  echo %PREFIX_ERR%Failed to write cluster ID to: "%CLUSTER_ID_FILE%"
  exit /b 1
)

echo %PREFIX_INFO%Cluster ID saved to: "%CLUSTER_ID_FILE%"

REM ---------------------------------------------------------------------------
REM STEP 3 - Format Storage Directory
REM ---------------------------------------------------------------------------
echo %PREFIX_INFO%Step 3/3 - Formatting Kafka storage directory...

"%KAFKA_HOME%\bin\windows\kafka-storage.bat" format -t "%KAFKA_CLUSTER_ID%" -c "%KAFKA_CONFIG%"
if NOT %ERRORLEVEL%==0 (
  echo %PREFIX_ERR%Storage format failed.
  exit /b 1
)

echo %PREFIX_INFO%Storage directory formatted successfully.
exit /b 0

This script is used to format the KRaft storage. It is a one-time task that must be executed manually before starting Kafka for the first time.

🚨 Do not execute the kafka-init.bat script at this stage. It should be executed later, just before starting the services.

Schema Registry

Install Schema Registry

• Download the latest Confluent release from the URL below.

https://packages.confluent.io/archive/

• Extract the Confluent .tar.gz file into the C:\kafka directory using the following steps. ‣ Create a folder named confluent inside the C:\kafka directory. ‣ Open PowerShell in Administrator mode, navigate to the downloads directory, and execute the following command.

tar -zxvf confluent-8.2.0.tar.gz -C C:\kafka\confluent --strip-components=1

🚨 Do not download the ZIP file, as some symbolic link files may be skipped during extraction.

Kafka UI

Install Kafka UI

• Download the latest Kafka UI JAR file from the URL below.

https://github.com/provectus/kafka-ui/releases

• Create a folder named kafka-ui inside the C:\kafka directory.
• Move the downloaded kafka-ui-api-v0.7.2.jar file into the C:\kafka\kafka-ui directory.

Configure Kafka UI

• Navigate to the C:\kafka\kafka-ui directory.
• Create a file named application.yml and add the following configurations.

server:
  port: 9000
kafka:
  clusters:
    - name: kafka-cluster
      bootstrapServers: localhost:9092
      schemaRegistry: http://localhost:8081
      kafkaConnect:
        - name: debezium-connect
          address: http://localhost:8083
# Enable read/write operations to allow creating topics and connectors from the UI.
auth:
  type: disabled
management:
  health:
    ldap:
      enabled: false

Create Service Start Scripts

• Navigate to the C:\kafka directory.
• Create the following batch (.bat) files to start each service.

‣ Create a batch file named start-kafka.bat and add the following content to it.

@echo off
setlocal

:: Start the Kafka server
echo Starting Kafka Broker...
call .\bin\windows\kafka-server-start.bat .\config\kraft\server.properties

pause

‣ Create a batch file named start-schema-registry.bat and add the following content to it.

@echo off
setlocal

:: Start the schema-registry
echo Starting Schema Registry...
call java -cp "confluent/share/java/schema-registry/*;confluent/share/java/confluent-common/*;confluent/share/java/rest-utils/*"  io.confluent.kafka.schemaregistry.rest.SchemaRegistryMain confluent/etc/schema-registry/schema-registry.properties

pause

‣ Create a batch file named start-kafka-connect.bat and add the following content to it.

@echo off
setlocal

:: Set Kafka Connect heap memory to 4GB
set KAFKA_HEAP_OPTS=-Xms4G -Xmx4G

:: Set JVM performance options
set KAFKA_JVM_PERFORMANCE_OPTS=-XX:+UseG1GC -XX:MaxGCPauseMillis=20 -XX:InitiatingHeapOccupancyPercent=35 -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=%%TEMP%%\

:: Start the Kafka server
echo Starting Kafka Connect...
call .\bin\windows\connect-distributed.bat .\config\connect-distributed.properties

pause

‣ Create a batch file named start-kafka-ui.bat and add the following content to it.

@echo off
setlocal

echo Starting Kafka UI...
call java -jar .\kafka-ui\kafka-ui-api-v0.7.2.jar --spring.profiles.active= --spring.config.additional-location=kafka-ui\application.yml

pause

Start the Services

Before starting the services, the KRaft storage must be formatted by executing the kafka-init.bat script.

Please note that this is a one-time task that must be executed manually before the Kafka service is started for the first time.

• Open PowerShell or CMD and navigate to the C:\kafka\ directory.
• Execute the following command.

.\kafka-init.bat

The following output will be displayed once the script has been executed successfully.

kafka-init.bat file execution output log

Common Issues

Java Installation Path Contains Spaces

If an error occurs because the Java installation path contains spaces, follow the steps below.

• Navigate to the C:\kafka\bin\windows directory.
• Open the kafka-run-class.bat file using Notepad or Notepad++.
• Go to the end of the file and locate the set COMMAND= line.
• Wrap %JAVA% with double quotes as shown below.

"%JAVA%"

• Execute kafka-init.bat again to verify the issue has been resolved.

Windows 11 WMIC Issue

In Windows 11, the wmic command may not work. Follow the steps below to update the kafka-server-start.bat file.

• Navigate to the C:\kafka\bin\windows directory.
• Open the kafka-server-start.bat file using Notepad or Notepad++.
• Remove the wmic command and the related IF ELSE condition shown in the image.
• Keep only the 64-bit OS set KAFKA_HEAP_OPTS command.

After editing the kafka-server-start.bat

Execute the following batch files in separate PowerShell windows to start the services in order.

.\start-kafka.bat
.\start-schema-registry.bat
.\start-kafka-connect.bat
.\start-kafka-ui.bat

Once all services are running, open the Kafka UI dashboard using the URL below.

http://localhost:9000

kafka-ui dashboard

Configure Kafka Connect MySQL Connector

Before creating the Kafka connector, a debezium_signal table must be created in the source database.

This table is used by Debezium to manage signaling operations and trigger connector actions such as:

• Adding new tables for CDC
• Triggering incremental snapshots
• Refreshing existing data without restarting the connector

Use the following query to create the table in the source database.

CREATE TABLE debezium_signal (
  id VARCHAR(42) PRIMARY KEY,
  type VARCHAR(32) NOT NULL,
  data VARCHAR(2048) NULL
);

The table must exist before starting the connector. Otherwise, the connector will fail to start.

Follow the steps below to create the Debezium MySQL source connector.

• Open the Kafka UI dashboard using the URL below.

http://localhost:9000

• Navigate to the Kafka Connect page from the left menu.
• Click the Create Connector button.
• Enter a connector name (example: source-db-connector).
• Add the following connector configurations in the Config section.

{
  "connector.class": "io.debezium.connector.mysql.MySqlConnector",
  "tasks.max": "1",

  "database.hostname": "localhost",
  "database.port": "3306",
  "database.user": "<db-username>",
  "database.password": "<db-password>",
  "database.server.id": "<mysql-server-id>",
  "database.include.list": "testdb",
  "database.connectionTimeZone": "UTC",

  "table.include.list": "testdb.customer,testdb.invoice",

  "topic.prefix": "core",
  "include.schema.changes": "false",
  "tombstones.on.delete": "true",

  "snapshot.mode": "initial",
  "snapshot.fetch.size": "10240",
  "snapshot.max.threads": "1",
  "snapshot.locking.mode": "none",

  "decimal.handling.mode": "precise",
  "time.precision.mode": "adaptive_time_microseconds",

  "schema.history.internal.kafka.topic": "schema-history.mysql",
  "schema.history.internal.kafka.bootstrap.servers": "localhost:9092",

  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "http://localhost:8081",
  "key.converter.schemas.cache.config": "1000",

  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter.schema.registry.url": "http://localhost:8081",
  "value.converter.schemas.cache.config": "1000",

  "max.batch.size": "8192",
  "max.queue.size": "32768",
  "max.queue.size.in.bytes": "134217728",
  "poll.interval.ms": "1000",

  "min.row.count.to.stream.results": "1000",
  "connect.keep.alive": "true",
  "connect.keep.alive.interval.ms": "60000",

  "signal.data.collection": "testdb.debezium_signal",
  "signal.enabled.channels": "source",
  "incremental.snapshot.enabled": "true",
  "incremental.snapshot.chunk.size": "1024",
  "heartbeat.interval.ms": "10000",

  "producer.override.max.request.size": "10485760",
  "producer.override.buffer.memory": "67108864",
  "producer.override.batch.size": "524288",
  "producer.override.linger.ms": "20",
  "producer.override.compression.type": "lz4",
  "producer.override.acks": "1"
}

• Update the database configurations in the above connector configurations.
• Click Submit to create the connector.

Once the connector is created successfully, Debezium will start reading database changes and publishing them to Kafka topics.

Topic data and connector progress can be viewed from the Topics page in Kafka UI.

kafka topics

Steps to Reset Data in Kafka Topics

Follow the steps below to delete all existing Kafka topic data, Kafka Connect data, and related information. This will completely clear Kafka and restart it as a fresh environment.

• Navigate to the C:\kafka directory.
• Create a batch file named create-meta-data.bat and add the following content to it.

@echo off
setlocal EnableExtensions EnableDelayedExpansion

set "KAFKA_HOME=C:\kafka"
set "CLUSTER_ID_FILE=%KAFKA_HOME%\config\kraft\.cluster_id"
set "META_DIR=%KAFKA_HOME%\data"
set "META_FILE=%META_DIR%\meta.properties"
set "NODE_ID=1"
set "VERSION=1"

REM --- Pre-checks ---
if not exist "%CLUSTER_ID_FILE%" (
  echo [ERROR] Cluster ID file not found: "%CLUSTER_ID_FILE%"
  exit /b 1
)

if not exist "%META_DIR%" (
  mkdir "%META_DIR%"
  if NOT !ERRORLEVEL!==0 (
    echo [ERROR] Failed to create directory: "%META_DIR%"
    exit /b 1
  )
)

REM --- Read cluster id (first line) ---
set "CLUSTER_ID="
for /f "usebackq delims=" %%A in ("%CLUSTER_ID_FILE%") do set "CLUSTER_ID=%%A"

if not defined CLUSTER_ID goto :cluster_id_empty
goto :cluster_id_ok

:cluster_id_empty
echo [ERROR] Cluster ID is empty (file: "%CLUSTER_ID_FILE%")
exit /b 1

:cluster_id_ok

REM --- Build timestamp like: Tue 05/05/2026 14:33:12.34 ---
for /f "delims=" %%T in ('powershell -NoProfile -Command "Get-Date"') do set "NOW=%%T"

REM --- Write meta.properties ---
(
  echo #
  echo #!NOW!
  echo node.id=%NODE_ID%
  echo version=%VERSION%
  echo cluster.id=!CLUSTER_ID!
) > "%META_FILE%"

if NOT !ERRORLEVEL!==0 (
  echo [ERROR] Failed to write: "%META_FILE%"
  exit /b 1
)

echo [INFO] Wrote "%META_FILE%"
exit /b 0

• Create a batch file named reset-kafka-data.bat and add the following content to it.

@echo off
setlocal EnableExtensions

set "KAFKA_HOME=C:\kafka"
set "DATA_DIR=%KAFKA_HOME%\data"
set "CREATE_META_BAT=%KAFKA_HOME%\create-meta-data.bat"

echo [INFO] Resetting Kafka data directory: "%DATA_DIR%"

REM Ensure data directory exists
if not exist "%DATA_DIR%" (
  echo [INFO] Data directory does not exist; creating: "%DATA_DIR%"
  mkdir "%DATA_DIR%"
  if NOT %ERRORLEVEL%==0 (
    echo [ERROR] Failed to create "%DATA_DIR%"
    exit /b 1
  )
)

REM Delete all contents under C:\kafka\data (but keep the directory itself)
echo [INFO] Deleting all contents in "%DATA_DIR%"...
del /q /f "%DATA_DIR%\*" >nul 2>&1
for /d %%D in ("%DATA_DIR%\*") do rd /s /q "%%D" >nul 2>&1

REM Basic verification that create-meta-data.bat exists
if not exist "%CREATE_META_BAT%" (
  echo [ERROR] "%CREATE_META_BAT%" not found.
  exit /b 1
)

REM Execute create-meta-data.bat from C:\kafka
echo [INFO] Running "%CREATE_META_BAT%"...
pushd "%KAFKA_HOME%" >nul
call "%CREATE_META_BAT%"
set "RC=%ERRORLEVEL%"
popd >nul

if NOT "%RC%"=="0" (
  echo [ERROR] create-meta-data.bat failed with exit code %RC%.
  exit /b %RC%
)

echo [INFO] Kafka data reset complete.
exit /b 0

• Stop all the services (kafka, schema-registry, kafka-connect and kafka-ui).
• Execute reset-kafka-data.bat file using the command below to delete all the Kafka data.

.\reset-kafka-data.bat

• Start all services sequentially. Then, use Kafka UI to verify whether the data has been deleted successfully.

Conclusion

At this stage, the Debezium CDC pipeline is fully configured and running successfully. Database changes from the source MySQL database will now be captured in real time and published to Kafka topics automatically.

The published topic data can be consumed in multiple ways depending on the requirement. One option is to create a backend consumer application using frameworks such as Spring Boot to listen to Kafka topics and process the events.

Another option is to use Kafka Connect sink connectors to directly synchronize the Kafka topic data into another database, search engine, or external system without writing custom application code.

This setup provides a scalable and flexible foundation for building real-time data streaming, event-driven systems, and data synchronization solutions.

That’s all. Cheers ✌️