Learn Debezium - Installation & Setting Up Connectors

#Introduction

In episode 0 you already started Kafka and Kafka Connect in KRaft mode. Episode 3 puts that runtime to real use: registering your first connector. You'll add a source database, create a dedicated user with minimal privileges, register the connector through the REST API, and read your first CDC event in the terminal.

Although the main example is MySQL, the end of this episode summarizes configurations for PostgreSQL, MongoDB, and SQL Server. With one clear working pattern, you can add a new data source simply by copying and adjusting the configuration.

#Setting Up the Source Database with Docker

Add a MySQL service to the existing docker-compose.yml. Debezium provides an example image that contains the inventory database complete with a debezium user that already has the correct privileges:

  mysql:
    image: quay.io/debezium/example-mysql:3.0
    ports:
      - "3306:3306"
    environment:
      MYSQL_ROOT_PASSWORD: dbz
    volumes:
      - mysql_data:/var/lib/mysql
 
volumes:
  mysql_data:

Restart the stack so the MySQL service comes up too:

docker compose up -d
docker compose ps

Make sure the three containers — kafka, connect, and mysql — are in the running state before continuing.

#Registering the MySQL Connector

Connectors are registered through the Kafka Connect REST API on port 8083. The configuration is sent as a JSON document to the /connectors endpoint:

curl -i -X POST -H "Accept: application/json" -H "Content-Type: application/json" \
  http://localhost:8083/connectors/ -d '{
    "name": "inventory-connector",
    "config": {
      "connector.class": "io.debezium.connector.mysql.MySqlConnector",
      "database.hostname": "mysql",
      "database.port": "3306",
      "database.user": "debezium",
      "database.password": "dbz",
      "database.server.id": "223344",
      "topic.prefix": "dbserver1",
      "database.include.list": "inventory",
      "schema.history.internal.kafka.topic": "schema-changes.inventory",
      "schema.history.internal.kafka.bootstrap.servers": "kafka:9092"
    }
  }'

A few properties deserve special attention:

• database.server.id must be unique and not match any other server in the MySQL cluster.
• topic.prefix becomes the prefix for all output topic names.
• schema.history.internal.kafka.topic stores the schema history so a MySQL without binlog schema can still be reconstructed.

Verify the connector status with:

curl -s http://localhost:8083/connectors/inventory-connector/status

The output must show "state": "RUNNING" for both the connector and its task. If it's still FAILED, read the connect container logs to find out why.

#Configuration for Other Sources

The same pattern applies to other databases — only the connector class and connection properties differ.

#PostgreSQL

PostgreSQL uses a replication slot and a publication. The important properties include plugin.name set to pgoutput, slot.name, and publication.name:

{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "database.hostname": "postgres",
  "database.port": "5432",
  "database.user": "postgres",
  "database.password": "postgres",
  "database.dbname": "postgres",
  "topic.prefix": "dbserver2",
  "schema.include.list": "public",
  "plugin.name": "pgoutput",
  "slot.name": "debezium_slot",
  "publication.name": "dbz_publication"
}

#MongoDB and SQL Server

For MongoDB, Debezium needs the replica set connection string and a topic.prefix. For SQL Server, just the host, port, database name, and the schema you want to capture:

{
  "connector.class": "io.debezium.connector.mongodb.MongoDbConnector",
  "mongodb.connection.string": "mongodb://mongo:27017/?replicaSet=rs0",
  "mongodb.user": "debezium",
  "mongodb.password": "dbz",
  "topic.prefix": "mongo1"
}
{
  "connector.class": "io.debezium.connector.sqlserver.SqlServerConnector",
  "database.hostname": "sqlserver",
  "database.port": "1433",
  "database.user": "sa",
  "database.password": "Password!1",
  "database.names": "testdb",
  "schema.include.list": "dbo",
  "topic.prefix": "sqlserver1"
}

Note: MongoDB requires replica set mode, not standalone, because change streams are only available on replica sets.

#The Required User Privileges

Minimum privileges keep the connector from becoming a security hole. For MySQL, the Debezium user needs the SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, and REPLICATION CLIENT grants:

Create the Debezium user in MySQL

CREATE USER 'debezium'@'%' IDENTIFIED BY 'dbz';
GRANT SELECT, RELOAD, SHOW DATABASES,
      REPLICATION SLAVE, REPLICATION CLIENT
  ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

For PostgreSQL, the user must have the REPLICATION and SELECT privileges on the target tables. Never use a root account for production connectors — the principle of least privilege applies to every database.

#Verifying the Output Topic and the First CDC Message

Once the connector is RUNNING, the initial snapshot fills the topic with events whose op is r. To prove streaming is working, add a new row:

docker exec -it mysql bash -c 'mysql -u root -pdbz inventory -e "INSERT INTO customers VALUES (1001, \"Budi\", \"Santoso\", \"budi@example.com\")"'

Then read the topic with a Kafka consumer:

docker exec -it kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka:9092 \
  --topic dbserver1.inventory.customers \
  --from-beginning \
  --property print.key=true

The events displayed have a key containing the row id and a value containing a payload object with op set to c. That means the installation succeeded: every change to the customers table now flows to Kafka in real time.

#Conclusion

Episode 3 took you from zero to your first CDC message: adding MySQL as the source, registering a connector through the REST API, understanding the configuration for four popular databases, arranging minimal privileges, and verifying events landing in the topic.

The key takeaways:

• Connectors are registered via POST /connectors with a JSON body.
• topic.prefix and database.server.id are required properties and must be unique.
• Each database has its own connection mechanism: pgoutput for PostgreSQL, replica set for MongoDB.
• Always create a Debezium user with minimal privileges.
• Verify CDC by inserting a new row and then reading the topic with the console consumer.

In the next episode 4 we'll dissect snapshot, streaming, and offset management — how Debezium fills in the initial data, how offsets are stored and recovered, snapshot chunking for large tables, and how to handle schema changes.