Change Data Capture from MySQL to Kafka

Setting Up Environment

Before we dive into the code, make sure you have a Kafka ecosystem set up on your MacOS machine using Docker. You can refer to our previous article on setting up a Kafka container using Docker for more information.

Installing the Debezium Connector

To use the Debezium MySQL CDC Source Connector, install it using Confluent Hub. Add the below entry inside the docker kafka-connect service under the command.

confluent-hub install --no-prompt debezium/debezium-connector-mysql:2.4.2

Simulating MySQL Source Database

Add a Docker container service to simulate a MySQL source database in the docker-compose.yml file.

# MySQL Source Database
  mysql_src:
    # Apple M1 Chip
    # platform: linux/amd64
    image: mysql:5.7
    container_name: mysql_src
    restart: always
    environment:
      MYSQL_ROOT_HOST: '%'
      MYSQL_DATABASE: sales
      MYSQL_ROOT_PASSWORD: Password1234
    ports:
      - 3308:3306
    volumes:
      - mysql_datadir:/var/lib/mysql
    networks:
      - kafka-network
    command:
    - --log-bin=binlog
    - --binlog-format=ROW
    - --server-id=1
    - --sql_mode=

Please also remember to add mysql_datadir under the volumes block.

Note:The MySQL connector uses a client library for accessing the binlog. The command "--log-bin=binlog", "--binlog-format=ROW", "--server-id=1" configures the MySQL database to start with binary logging.

Create Source Table

Create a source table with some data in the MySQL database for our demo:

CREATE TABLE consultants(
    id INT AUTO_INCREMENT PRIMARY KEY,
    first_name VARCHAR(50),
    last_name VARCHAR(50),
    email VARCHAR(50),
    rate DECIMAL(8,2),
    status VARCHAR(20),
    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP
);

INSERT INTO consultants(first_name, last_name, email, rate, status) VALUES 
    ('John', 'Doe', 'john.doe@gmail.com', 3000.00, 'perm'),
    ('Tom', 'Hanks', 'tom.hanks@yahoo.com', 3500.75, 'contract'),
    ('Jane', 'Doe', 'jane.doe@moneybank.com', 3500.75, 'perm'),
    ('Duke', 'Johnson', 'duke@hello.com', 4500.25, 'contract'),
    ('Peter', 'Parker', 'peter@gmail.com', 4500.25, 'contract'),
    ('Rick', 'Nice', 'rick@gmail.com', 4900, 'contract'),
    ('Tommy', 'Hill', 'tommy@gmail.com', 4100, 'perm'),
    ('Jill', 'Stone', 'jill@gmail.com', 4250.50, 'contract'),
    ('Honey', 'Bee', 'honey@gmail.com', 3200, 'perm'),
    ('Bell', 'Doe', 'bell@gmail.com', 34000, 'contract')
;

SELECT * FROM consultants;

Kafka Connect Configuration

Create a configuration file for the Debezium MySQL CDC Source Connector:

{
    "name": "src-mysql-debezium-cdc-consultants",
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "database.hostname": "host.docker.internal",
    "database.port": "3308",
    "database.user": "root",
    "database.password": "Password1234",
    "database.server.id": "101",
    "table.include.list": "sales.consultants",
    "include.schema.changes": "false",
    "decimal.handling.mode": "string",
    "time.precision.mode": "connect",
    "topic.prefix": "mysql_src",
    "schema.history.internal.kafka.topic": "dbhistory.mysql-sales",
    "schema.history.internal.kafka.bootstrap.servers": "broker:29092",
    "tasks.max": "1"
}

Once the Kafka Connector is up & running, let us now look into the Kafka Topic.

Capture Changed Data from Database

Modify some data in the MySQL database to observe the CDC from the database to a Kafka topic.

INSERT INTO consultants(first_name, last_name, email, rate, status) VALUES ('Saurav', 'Mitra', 'saurav.karate@gmail.com', 5000.00, 'perm');

UPDATE consultants set rate=6500.00 where email='saurav.karate@gmail.com';

INSERT INTO consultants(first_name, last_name, email, rate, status) VALUES ('Tim', 'Smith', 'tim.smith@freelance.com', 3500.70, 'contract');

DELETE from consultants where email='tim.smith@freelance.com';

INSERT INTO consultants(first_name, last_name, email, rate, status) VALUES 
    ('Shane', 'Wilson', 'shane.wilson@freelance.com', 5000.00, 'perm'),
    ('John', 'Sinha', 'john.sinha@freelance.com', 9000.00, 'contract')
;

We can now see, that the database events have been streamed in near real-time to the Kafka Topic.

tombstones.on.delete: true is the default setting.
After a source record is deleted, a delete operation is represented by a delete event and a subsequent tombstone event in Kafka Topic. The tombstone event allows Kafka to completely delete all events that pertain to the key of the deleted row in case log compaction is enabled for the topic.

In this article, we've demonstrated how to capture change data from a MySQL database and write it to a Kafka topic using Debezium's MySQL CDC Source Connector. With this setup, you can now process changes in real-time and integrate them with your existing data pipelines.