← Back to Blog

What Is Change Data Capture (CDC)? CDC With Debezium

cdc kafka debezium kafkaconnect

Working with database systems has always been a difficult, hair-graying job. I recently had the chance to work with a new solution for detecting changes in one database and replicating that data to another database, of the same or a different type.

To handle this problem, I used a solution with the name in the title — Change Data Capture, or CDC for short.

What is Change Data Capture?

Exactly as the name suggests, capturing data changes — this is a technique used to capture changes to data stored in a database. Being able to capture data changes helps us solve quite a few data-processing problems, which we'll explore in the next section.

There are many different ways to capture these data changes — the most primitive is using the built-in TRIGGER mechanism supported in databases to catch UPDATE, INSERT, DELETE, etc. actions. Or, more conveniently, we can use tools built for this purpose, and the most prominent one right now is Debezium.

Benefits of CDC

The first benefit everyone probably notices is replicating data to other systems. Talking about this benefit, some of you might say: databases already support replica mechanisms, so why bother with an external tool that's a headache and less stable? Okay, true! If you only need to replicate data between databases of the same type (MySQL => MySQL, Mongo => Mongo), using the database's own feature is the best option. But if you want to replicate data from MySQL to MongoDB, or from MySQL to PostgreSQL, there's no such mechanism. In this case, CDC sits in the middle to detect changes in the Database being watched and process the data, then you can use code to process and push data into the system that needs to replicate the data.

Another benefit that's no less important is the ability to back up data. Data-change events get stored, so if by any chance your database gets dropped at 9am, you can grab the 3am backup and apply the changes recorded from 3am to 9am. In theory, if you don't miss any event, your data will be fully restored to how it was before the drop. Pretty great, right 😃))))

Continuing with the first benefit — after replicating data to another system, we can use this system for testing instead of interacting directly with the real database system. It's not uncommon during testing for Developers to run queries that take minutes to process, or worse, cause the system to lock up. This issue can range from reducing system performance to causing a crash entirely. CDC is also a way to help us reduce the chances of this happening.

CDC also supports handling some specific problems for each system, or processing Big Data — if you've applied CDC to these kinds of problems, share it with me below.

Debezium — a CDC Tool

Talking theory forever without an example isn't very useful, so I'll introduce a tool I've had time to work with and found quite good: Debezium. At its core, Debezium uses Kafka to generate messages corresponding to data-change events. Debezium uses Connectors to connect to database systems and capture changes — currently Debezium 1.9 supports MySQL, PostgreSQL, MongoDB, Oracle, SQL Server, DB2, Cassandra, Vitess. You can check the guide for each connector in the official docs: debezium.io

Here's Debezium's model: first there's a source DB — where we watch for data changes. Kafka Connect plays the role of detecting changes and pushing events into Kafka Apache. From there, data can get pushed out to sinks depending on the use case.

In this section I'll describe the setup steps for MySQL, the type of database most people probably work with the most.

For MySQL, Debezium relies on the binlog to detect data changes, so for the system you need to monitor, you need to enable this binlog feature, and make sure the user used to connect has permission to read the binlog.

I work with Kubernetes a lot, so I'll walk through setting up this tool on K8s — for other environments like VM or Docker, the components involved are basically similar.

Debezium running on K8s uses the Strimzi Operator (a Kafka Operator). First, let's create a dedicated namespace for this application:

kubectl create ns debezium-example

Then we need to install the Strimzi Operator

curl -sL https://github.com/operator-framework/operator-lifecycle-manager/releases/download/v0.20.0/install.sh | bash -s v0.20.0

Create a Secret for the demo database

cat << EOF | kubectl create -n debezium-example -f
apiVersion: v1
kind: Secret
metadata:
  name: debezium-secret
  namespace: debezium-example
type: Opaque
data:
  username: ZGViZXppdW0=
  password: ZGJ6
EOF

Create a User and grant permissions for Debezium

cat << EOF | kubectl create -n debezium-example -f
apiVersion: rbac.authorization.k8s.io/v1
kind: Role
metadata:
  name: connector-configuration-role
  namespace: debezium-example
rules:
- apiGroups: [""]
  resources: ["secrets"]
  resourceNames: ["debezium-secret"]
  verbs: ["get"]
EOF
$ cat << EOF | kubectl create -n debezium-example -f
apiVersion: rbac.authorization.k8s.io/v1
kind: RoleBinding
metadata:
  name: connector-configuration-role-binding
  namespace: debezium-example
subjects:
- kind: ServiceAccount
  name: debezium-connect-cluster-connect
  namespace: debezium-example
roleRef:
  kind: Role
  name: connector-configuration-role
  apiGroup: rbac.authorization.k8s.io
EOF

Now for the important part — we'll spin up a Kafka cluster to store change events. The following config creates 1 kafka pod for 1 broker and 1 zookeeper pod.

$ cat << EOF | kubectl create -n debezium-example -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: Kafka
metadata:
  name: debezium-cluster
spec:
  kafka:
    replicas: 1
    listeners:
      - name: plain
        port: 9092
        type: internal
        tls: false
      - name: tls
        port: 9093
        type: internal
        tls: true
        authentication:
          type: tls
      - name: external
        port: 9094
        type: nodeport
        tls: false
    storage:
      type: jbod
      volumes:
      - id: 0
        type: persistent-claim
        size: 100Gi
        deleteClaim: false
    config:
      offsets.topic.replication.factor: 1
      transaction.state.log.replication.factor: 1
      transaction.state.log.min.isr: 1
      default.replication.factor: 1
      min.insync.replicas: 1
  zookeeper:
    replicas: 1
    storage:
      type: persistent-claim
      size: 100Gi
      deleteClaim: false
  entityOperator:
    topicOperator: {}
    userOperator: {}
EOF

Next, let's deploy a MySQL database to test with — this DB's user and pass are mysqluser - mysqlpw

cat << EOF | kubectl create -n debezium-example -f -
apiVersion: v1
kind: Service
metadata:
  name: mysql
spec:
  ports:
  - port: 3306
  selector:
    app: mysql
  clusterIP: None
---
apiVersion: apps/v1
kind: Deployment
metadata:
  name: mysql
spec:
  selector:
    matchLabels:
      app: mysql
  strategy:
    type: Recreate
  template:
    metadata:
      labels:
        app: mysql
    spec:
      containers:
      - image: quay.io/debezium/example-mysql:1.9
        name: mysql
        env:
        - name: MYSQL_ROOT_PASSWORD
          value: debezium
        - name: MYSQL_USER
          value: mysqluser
        - name: MYSQL_PASSWORD
          value: mysqlpw
        ports:
        - containerPort: 3306
          name: mysql
EOF

Now we'll deploy the components that connect to MySQL and detect changes. First we need to create a KafkaConnect to handle the change-detection job:

$ cat << EOF | kubectl create -n debezium-example -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: debezium-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 3.1.0
  replicas: 1
  bootstrapServers: debezium-cluster-kafka-bootstrap:9092
  config:
    config.providers: secrets
    config.providers.secrets.class: io.strimzi.kafka.KubernetesSecretConfigProvider
    group.id: connect-cluster
    offset.storage.topic: connect-cluster-offsets
    config.storage.topic: connect-cluster-configs
    status.storage.topic: connect-cluster-status
    # -1 means it will use the default replication factor configured in the broker
    config.storage.replication.factor: -1
    offset.storage.replication.factor: -1
    status.storage.replication.factor: -1
  build:
    output:
      type: docker
      image: 10.110.154.103/debezium-connect-mysql:latest
    plugins:
      - name: debezium-mysql-connector
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/{debezium-version}/debezium-connector-mysql-{debezium-version}-plugin.tar.gz
EOF

Then we deploy a KafkaConnector to connect to MySQL, attached to the KafkaConnect created above

cat << EOF | kubectl create -n debezium-example -f -
apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnector
metadata:
  name: debezium-connector-mysql
  labels:
    strimzi.io/cluster: debezium-connect-cluster
spec:
  class: io.debezium.connector.mysql.MySqlConnector
  tasksMax: 1
  config:
    tasks.max: 1
    database.hostname: mysql
    database.port: 3306
    database.user: ${secrets:debezium-example/debezium-secret:username}
    database.password: ${secrets:debezium-example/debezium-secret:password}
    database.server.id: 184054
    database.server.name: mysql
    database.include.list: inventory
    database.history.kafka.bootstrap.servers: debezium-cluster-kafka-bootstrap:9092
    database.history.kafka.topic: schema-changes.inventory
EOF

So the setup is now complete and we can watch for changes in the database. Run this command to listen to messages in kafka

kubectl run -n debezium-example -it --rm --image=quay.io/debezium/tooling:1.2  --restart=Never watcher -- kcat -b debezium-cluster-kafka-bootstrap:9092 -C -o beginning -t mysql.inventory.customers

Open another terminal — now let's access the DB and add a record to test:

kubectl run -n debezium-example -it --rm --image=mysql:8.0 --restart=Never --env MYSQL_ROOT_PASSWORD=debezium mysqlterm -- mysql -hmysql -P3306 -uroot -pdebezium

Add a record:

sql> update customers set first_name="Sally Marie" where id=1001;

If you see a JSON message like this, the setup succeeded:

{
...
  "payload": {
    "before": {
      "id": 1001,
      "first_name": "Sally",
      "last_name": "Thomas",
      "email": "sally.thomas@acme.com"
    },
    "after": {
      "id": 1001,
      "first_name": "Sally Marie",
      "last_name": "Thomas",
      "email": "sally.thomas@acme.com"
    },
    "source": {
      "version": "{debezium-version}",
      "connector": "mysql",
      "name": "mysql",
      "ts_ms": 1646300467000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": null,
      "table": "customers",
      "server_id": 223344,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 401,
      "row": 0,
      "thread": null,
      "query": null
    },
    "op": "u",
    "ts_ms": 1646300467746,
    "transaction": null
  }
}

The event message has 3 main sections: source (where the data came from, e.g. which binlog file), before (the data before the change), and after (the data after the change). From the before and after sections, we can tell exactly how the data changed (updated, deleted, or inserted)

Debezium can also do plenty of other things, like automatically replicating changes between 2 different DBs, etc.

Wrapping up

I ran into quite a few issues during setup :v, partly because I didn't have much experience with Kafka yet. If you run into an error setting this up, you're not alone 😃))) Feel free to comment whatever error you're seeing and I'll help. Hope this post gave you some more knowledge about Change Data Capture.

A few sources I referenced:

If you're running into technical challenges or need help with systems, DevOps tools, I'm confident I can help. Get in touch at hoangviet.io.vn

Thank you all! Have a nice day!

Have thoughts on this?

I'd love to hear your perspective. Send me a message.

Get in touch