{{announcement.body}}
{{announcement.title}}

Change Data Capture With Debezium: A Simple How-To, Part 1

DZone 's Guide to

Change Data Capture With Debezium: A Simple How-To, Part 1

This is a simple how-to on how to build out a change data capture solution using Debezium within an OpenShift environment.

· Integration Zone ·
Free Resource

One question always comes up as organizations moving towards being cloud-native, twelve-factor, and stateless: How do you get an organization’s data to these new applications? There are many different patterns out there, but one pattern we will look at today is change data capture. This post is a simple how-to on how to build out a change data capture solution using Debezium within an OpenShift environment. Future posts will also add to this and add additional capabilities.

What Is Change Data Capture?

Another Red Hatter, Sadhana Nandakumar, sums it up well in one of her posts around change data capture:

"Change data capture (CDC) is a pattern that enables database changes to be monitored and propagated to downstream systems. It is an effective way of enabling reliable microservices integration and solving typical challenges, such as gradually extracting microservices from existing monoliths."

This pattern lets data become distributed amongst teams, where each team can self-manage their own data while still keeping up-to-date with the original source of data. There are also other patterns, such as Command Query Responsibility Segregation (CQRS), which build on this idea.

What Is Debezium?

Debezium is an open source technology, supported by Red Hat as part of Red Hat Integration, which allows database row-level changes to be captured as events and published to Apache Kafka topics. Debezium connectors are based on the popular Apache Kafka Connect API and can be deployed within Red Hat AMQ Streams Kafka clusters.

Application Overview

The application we will use as our "monolith" is a Spring Boot application that uses a MySQL database as its back end. The application itself has adopted the Event Sourcing and Outbox patterns. This means that the application maintains a separate table within the database consisting of domain events. It is this table that we need to monitor for changes to publish into our Kafka topics. In this example, there is a table called outbox_events that looks like this:

Java
 




xxxxxxxxxx
1
10


 
1
+-----------------+--------------+------+-----+---------+----------------+
2
| Field           | Type         | Null | Key | Default | Extra          |
3
+-----------------+--------------+------+-----+---------+----------------+
4
| event_id        | bigint(20)   | NO   | PRI | NULL    | auto_increment |
5
| aggregate_id    | varchar(255) | NO   |     | NULL    |                |
6
| aggregate_type  | varchar(255) | NO   |     | NULL    |                |
7
| event_timestamp | datetime(6)  | NO   |     | NULL    |                |
8
| event_type      | varchar(255) | NO   |     | NULL    |                |
9
| payload         | json         | YES  |     | NULL    |                |
10
+-----------------+--------------+------+-----+---------+----------------+


Setting up the Database

The Debezium documentation has a section on how to set up the Debezium connector to work with a MySQL database. We need to follow that documentation but in a container-native way since we will run everything on Red Hat OpenShift. There are many different ways to accomplish this task, but I will describe the way I decided to do it.

Create an OpenShift Project

The first thing we need to do is log into our OpenShift cluster. In my example, I use OpenShift 4.3. The database setup does not require cluster admin privileges, so any normal user will work fine:

Java
 




x


 
1
$ oc login <CLUSTER_API_URL>


Next, let’s create a project to host our work:

Java
 




xxxxxxxxxx
1


 
1
$ oc new-project debezium-demo


Create the MySQL Configuration

From , the first thing we need to do is enable the binlog, GTIDs, and query log events. This is typically done in the MySQL configuration file, usually located in /etc/my.cnf. In our case, we will use Red Hat’s MySQL 8.0 container image. This image is already deployed in most OpenShift installations in the openshift namespace under the mysql:8.0 tag. The source of this image comes from registry.redhat.io/rhscl/mysql-80-rhel7:latest.

According to the container image documentation, the default configuration file is at /etc/my.cnf, but there is an environment variable, MYSQL_DEFAULTS_FILE, that can be used to override its location. MySQL configuration also lets one configuration file include other configuration files, so we will create a new configuration file that first includes the default configuration and then overrides some of that configuration to enable the required Debezium configuration.

We’ll do this by first creating a configuration file containing our configuration. We’ll call this file my-debezium.cnf:

Java
 




xxxxxxxxxx
1
12


 
1
!include /etc/my.cnf
2
 
          
3
[mysqld]
4
server-id = 223344
5
server_id = 223344
6
log_bin = ON
7
binlog_format = ROW
8
binlog_row_image = full
9
binlog_rows_query_log_events = ON
10
expire_logs_days = 10
11
gtid_mode = ON
12
enforce_gtid_consistency = ON


Now that our MySQL configuration file is created, let's create it as a ConfigMap within our OpenShift project:

Java
 




xxxxxxxxxx
1


 
1
$ oc create configmap db-config --from-file=my-debezium.cnf


Create a MySQL User

The next part of the Debezium MySQL configuration is to create a MySQL user for the connector. We will follow the same pattern that we did for the configuration by creating a file containing the needed SQL. This initdb.sql file will create a user with the ID debezium and password debezium:

Java
 




xxxxxxxxxx
1


 
1
CREATE USER IF NOT EXISTS 'debezium'@'%' IDENTIFIED WITH mysql_native_password BY 'debezium';
2
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
3
FLUSH PRIVILEGES;


Note: In a real production environment, we want to choose usernames and passwords more carefully, as well as only allowing the debezium user access to the tables it will monitor.

Now create a ConfigMap within our OpenShift project:

Java
 




xxxxxxxxxx
1


 
1
$ oc create configmap db-init --from-file=initdb.sql


The last piece of the configuration is to create an OpenShift Secret to hold onto our database credentials. This Secret will be used by our database as well as the application that connects to the database. For simplicity, we will use music as our database name, username, password, and admin password:

Java
 




xxxxxxxxxx
1


 
1
$ oc create secret generic db-creds --from-literal=database-name=music --from-literal=database-password=music --from-literal=database-user=music --from-literal=database-admin-password=music
2
 
          


Note (again): In a real production environment, we want to choose usernames and passwords more carefully.

Deploy MySQL

The last part is to create the database and point it to our two configurations. OpenShift allows us to take the ConfigMaps we created and mount them as files within the container filesystem. We can then use environment variables to change the behavior of the MySQL container image. Let’s create a descriptor YAML file, mysql.yml, for our database DeploymentConfig and Service:

Java
 




xxxxxxxxxx
1
132


 
1
kind: DeploymentConfig
2
apiVersion: apps.openshift.io/v1
3
metadata:
4
  name: spring-music-db
5
  labels:
6
    application: spring-music
7
    app: spring-music
8
    app.kubernetes.io/part-of: spring-music
9
    app.openshift.io/runtime: mysql-database
10
spec:
11
  replicas: 1
12
  strategy:
13
    type: Recreate
14
    recreateParams:
15
      post:
16
        failurePolicy: Abort
17
        execNewPod:
18
          command:
19
            - /bin/sh
20
            - '-c'
21
            - sleep 10 && MYSQL_PWD="$MYSQL_ROOT_PASSWORD" $MYSQL_PREFIX/bin/mysql -h $SPRING_MUSIC_DB_SERVICE_HOST -u root < /config/initdb.d/initdb.sql
22
          containerName: spring-music-db
23
          volumes:
24
            - db-init
25
  selector:
26
    name: spring-music-db
27
  template:
28
    metadata:
29
      name: spring-music-db
30
      labels:
31
        name: spring-music-db
32
    spec:
33
      volumes:
34
        - name: db-data
35
          emptyDir: {}
36
        - name: db-init
37
          configMap:
38
            name: db-init
39
        - name: db-config
40
          configMap:
41
            name: db-config
42
      containers:
43
        - env:
44
            - name: MYSQL_DEFAULTS_FILE
45
              value: /config/configdb.d/my-debezium.cnf
46
            - name: MYSQL_USER
47
              valueFrom:
48
                secretKeyRef:
49
                  name: db-creds
50
                  key: database-user
51
            - name: MYSQL_PASSWORD
52
              valueFrom:
53
                secretKeyRef:
54
                  name: db-creds
55
                  key: database-password
56
            - name: MYSQL_DATABASE
57
              valueFrom:
58
                secretKeyRef:
59
                  name: db-creds
60
                  key: database-name
61
            - name: MYSQL_ROOT_PASSWORD
62
              valueFrom:
63
                secretKeyRef:
64
                  name: db-creds
65
                  key: database-admin-password
66
          name: spring-music-db
67
          image: ' '
68
          imagePullPolicy: IfNotPresent
69
          volumeMounts:
70
            - name: db-data
71
              mountPath: /var/lib/mysql/data
72
            - name: db-init
73
              mountPath: /config/initdb.d
74
            - name: db-config
75
              mountPath: /config/configdb.d
76
          ports:
77
            - containerPort: 3306
78
              protocol: TCP
79
          livenessProbe:
80
            failureThreshold: 3
81
            initialDelaySeconds: 30
82
            periodSeconds: 10
83
            successThreshold: 1
84
            tcpSocket:
85
              port: 3306
86
            timeoutSeconds: 1
87
          readinessProbe:
88
            exec:
89
              command:
90
                - /bin/sh
91
                - -i
92
                - -c
93
                - MYSQL_PWD="$MYSQL_PASSWORD" mysql -h 127.0.0.1 -u $MYSQL_USER -D $MYSQL_DATABASE -e 'SELECT 1'
94
            failureThreshold: 3
95
            initialDelaySeconds: 5
96
            periodSeconds: 10
97
            successThreshold: 1
98
            timeoutSeconds: 1
99
          resources:
100
            limits:
101
              memory: 512Mi
102
          securityContext:
103
            privileged: false
104
  triggers:
105
    - type: ConfigChange
106
    - type: ImageChange
107
      imageChangeParams:
108
        automatic: true
109
        containerNames:
110
          - spring-music-db
111
        from:
112
          kind: ImageStreamTag
113
          name: mysql:8.0
114
          namespace: openshift
115
---
116
kind: Service
117
apiVersion: v1
118
metadata:
119
  name: spring-music-db
120
  labels:
121
    application: spring-music
122
    app: spring-music
123
  annotations:
124
    template.openshift.io/expose-uri: mysql://{.spec.clusterIP}:{.spec.ports[?(.name=="mysql")].port}
125
spec:
126
  ports:
127
    - name: mysql
128
      port: 3306
129
      protocol: TCP
130
      targetPort: 3306
131
  selector:
132
    name: spring-music-db


From this DeploymentConfig, you can see that we mount our db-init and db-config ConfigMaps as volumes on the container filesystem inside the /config directory on lines 72-75:

Java
 




xxxxxxxxxx
1


 
1
volumeMounts:
2
  - name: db-data
3
    mountPath: /var/lib/mysql/data
4
  - name: db-init
5
    mountPath: /config/initdb.d
6
  - name: db-config
7
    mountPath: /config/configdb.d


The /config/configdb.d/my-debezium.cnf file is also set as the value for the MYSQL_DEFAULTS_FILE environment variable on lines 44-45:

Java
 




xxxxxxxxxx
1


 
1
- env:
2
  - name: MYSQL_DEFAULTS_FILE
3
    value: /config/configdb.d/my-debezium.cnf


The database initialization script from the db-init ConfigMap is executed as a post lifecycle hook on lines 15-24:

Java
 




xxxxxxxxxx
1
10


 
1
post:
2
  failurePolicy: Abort
3
  execNewPod:
4
    command:
5
      - /bin/sh
6
      - '-c'
7
      - sleep 10 && MYSQL_PWD="$MYSQL_ROOT_PASSWORD" $MYSQL_PREFIX/bin/mysql -h $SPRING_MUSIC_DB_SERVICE_HOST -u root < /config/initdb.d/initdb.sql
8
    containerName: spring-music-db
9
    volumes:
10
      - db-init


Our MySQL instance here is ephemeral, so whenever a new container instance is created the script will execute in a sidecar container within the pod.

Now create the resources and wait for the database pod to start:

Java
 




xxxxxxxxxx
1


 
1
$ oc create -f mysql.yml


Starting the Application

Now that our database is up and running we can start the application. Let's go to the OpenShift web console and then to the Developer perspective's Topology view, as shown in Figure 1.

Navigate to Developer Perspective

Figure 1: Navigate to Developer Perspective

Then click the +Add button, followed by the Container Image tile, as shown in Figure 2.

Add new container image

Figure 2: Add new container image

Fill in the image name with quay.io/edeandrea/spring-music:latest and then click the search button, as shown in Figure 3.

Load container image

Figure 3: Load container image

Then fill out the rest of the information from Figures 4 and 5 below, making sure to add the correct labels and environment variables by clicking the links at the bottom with the sentence "Click on the names to access advanced options for Routing, Deployment, Scaling, Resource Limits, and Labels."

The fields and values should be filled out as follows:

  • Application: spring-music
  • Name: spring-music
  • Deployment Config: selected
  • Create a route to the application: checked
  • Labels
    • app.openshift.io/runtime=spring
  • Deployment
    • Auto deploy when new image is available: checked
    • Auto deploy when deployment configuration changes: checked
    • Environment Variables
      • Add from Config Map or Secret
        • NAME: SPRING_DATASOURCE_USERNAME
        • VALUE: From Secret db-creds field database-user
      • Add from Config Map or Secret
        • NAME: SPRING_DATASOURCE_PASSWORD
        • VALUE: from Secret db-creds field database-password
      • Add Value
        • NAME: SPRING_DATASOURCE_URL
        • VALUE: jdbc:mysql://spring-music-db/music

Complete application details, Part 1

Figure 4: Complete application details, Part 1


Complete application details, Part 2

Figure 5: Complete application details, Part 2

Once done, click the Create button.

Back in the Topology view, you should see the application spin up. Once it is surrounded by the blue ring, click the route button on the top-right corner of the application icon, as shown in Figure 6.

Launch application UI

Figure 6: Launch application UI

This will launch the application. Feel free to play around with it if you'd like. Try deleting an album.

Deploy AMQ Streams

Now that our database and application are up and running let’s deploy our AMQ Streams cluster. First, we need to install the AMQ Streams Operator into the cluster from the OperatorHub. To do this you need cluster admin privileges for your OpenShift cluster. Log in to the web console as a cluster admin, then on the left expand OperatorHub, search for AMQ Streams, and select Red Hat Integration - AMQ Streams, as shown in Figure 7.


Figure 7: Find AMQ Streams Operator

On the installation screen, click the Install button, as shown in Figure 8.

Install AMQ Streams Operator

Figure 8: Install AMQ Streams Operator

On the Create Operator Subscription page, leave the defaults and click Subscribe, as shown in Figure 9. This action will install the Operator for all of the projects in the cluster.

Create Operator subscription

Figure 9: Create Operator subscription

You'll then be brought to the Installed Operators screen. Sit tight and wait for the Red Hat Integration - AMQ Streams Operator to show up with Succeeded status, as shown in Figure 10. It shouldn't take more than a minute or two.

Wait for the Operator to provision

Figure 10: Wait for the Operator to provision

Now let's create our Kafka cluster. Click on the Red Hat Integration - AMQ Streams label to get to the main AMQ Streams Operator page. Then under Provided APIs, click the Create Instance label in the Kafka section, as shown in Figure 11.

Create Kafka instance

Figure 11: Create Kafka instance

The Create Kafka YAML editor will then come up. Remove everything that's there, paste in the following, and click the Create button at the bottom of the screen:

Java
 




xxxxxxxxxx
1
52


 
1
kind: Kafka
2
apiVersion: kafka.strimzi.io/v1beta1
3
metadata:
4
  name: db-events
5
  namespace: debezium-demo
6
  labels:
7
    app: spring-music-cdc
8
    template: spring-music-cdc
9
    app.kubernetes.io/part-of: spring-music-cdc
10
spec:
11
  kafka:
12
    replicas: 3
13
    listeners:
14
      plain: {}
15
    jvmOptions:
16
      gcLoggingEnabled: false
17
    config:
18
      auto.create.topics.enable: "true"
19
      num.partitions: 1
20
      offsets.topic.replication.factor: 3
21
      default.replication.factor: 3
22
      transaction.state.log.replication.factor: 3
23
      transaction.state.log.min.isr: 2
24
    storage:
25
      type: persistent-claim
26
      size: 100Gi
27
      deleteClaim: true
28
    template:
29
      statefulset:
30
        metadata:
31
          labels:
32
            app.kubernetes.io/part-of: spring-music-cdc
33
            app: spring-music-cdc
34
            template: spring-music-cdc
35
          annotations:
36
            app.openshift.io/connects-to: db-events-zookeeper
37
  zookeeper:
38
    replicas: 3
39
    storage:
40
      type: persistent-claim
41
      size: 100Gi
42
      deleteClaim: true
43
    template:
44
      statefulset:
45
        metadata:
46
          labels:
47
            app.kubernetes.io/part-of: spring-music-cdc
48
            app: spring-music-cdc
49
            template: spring-music-cdc
50
  entityOperator:
51
    topicOperator: {}
52
    userOperator: {}


This action will deploy a three-node Kafka cluster along with a three-node Zookeeper cluster. It will also turn down the JVM's garbage collection logging so that if we need to look at the logs in any of the Kafka broker pods they won’t be polluted with tons of garbage collection debug logs. Both the Kafka and Zookeeper brokers are backed by persistent storage, so the data will survive a broker and cluster restart.

Wait a few minutes for OpenShift to spin everything up. You can switch to the OpenShift Developer perspective’s Topology view by clicking what is shown in Figure 12.

Switch to developer perspective

Figure 12: Switch to developer perspective

Once the db-events-entity-operator, db-events-kafka, and db-events-zookeeper items all show up with a blue ring around them, as shown in Figure 13, you are done.

Wait for Kafka deployment

Figure 13: Wait for Kafka deployment

Deploy Kafka Connect

Debezium runs inside a Kafka Connect cluster, so that means we need a container image with both Kafka Connect and the Debezium libraries together. The easiest way to do this is to create your own container image from the Kafka Connect base image. What follows are the steps needed to do this. I also already created an image you can use, so feel free to skip this sub-section if you would like and use the image at quay.io/edeandrea/kafka-connect-debezium-mysql:amq-streams-1.4.0-dbz-1.1.0.Final instead.

Building Your Own Kafka Connect Image

To build your own Kafka Connect image:

  1. Create a directory on your local computer (i.e., debezium-connect-image) and then cd into that directory.
  2. Create a directory inside called plugins.
  3. Download the Debezium MySQL connector from the Debezium Releases page.

Note: This post was written using the 1.1.0.Final version of the MySQL connector, but whatever the latest version listed should do fine.

  1. Unpackage the downloaded file into the plugins directory.
  2. Create a Dockerfile at the root (i.e., debezium-connect-image) directory with the following contents (you'll need an account on registry.redhat.ioand to log into the registry on your machine in order to pull the AMQ Streams image):
    Java
     




    xxxxxxxxxx
    1


     
    1
    FROM registry.redhat.io/amq7/amq-streams-kafka-24-rhel7:1.4.0
    2
    USER root:root
    3
    COPY ./plugins/ /opt/kafka/plugins
    4
    USER jboss:jboss


  3. Your directory tree should now look like what's shown in Figure 14.
    Contents of Kafka Connect image
    Figure 14: Contents of Kafka Connect image
  4. Build or tag the image using your favorite tool (i.e., Docker/Buildah/etc.) and push it to your registry of choice.

Create Kafka Connect Credentials

Before we create the KafkaConnect cluster there is one small thing we need to take care of. The Debezium connector requires a connection to the database. Rather than hard-coding the credentials into the configuration, let’s instead create an OpenShift Secret that contains credentials that can then be mounted into the KafkaConnect pods.

On your local filesystem, create a file called connector.properties. The contents of this file should be:

Java
 




xxxxxxxxxx
1


 
1
dbUsername=debezium
2
dbPassword=debezium


Now let’s create the OpenShift Secret:

Java
 




xxxxxxxxxx
1


 
1
$ oc create secret generic db-connector-creds --from-file=connector.properties


Deploy the Kafka Connect image

Back in the OpenShift console go back to the Administrator perspective and go into Installed Operators, then click on the Red Hat Integration - AMQ Streams operator, as shown in Figure 15.

Installed Operators

Figure 15: Installed Operators


Then under Provided APIs, click the Create Instance label in the Kafka Connect section, as shown in Figure 16.

Create Kafka Connect instance

Figure 16: Create Kafka Connect instance


The Create KafkaConnect YAML editor will then come up. Remove everything that's there, paste in the following, and click the Create button at the bottom of the screen:

Java
 




xxxxxxxxxx
1
40


 
1
kind: KafkaConnect
2
apiVersion: kafka.strimzi.io/v1beta1
3
metadata:
4
  name: db-events
5
  namespace: debezium-demo
6
  labels:
7
    app: spring-music-cdc
8
    template: spring-music-cdc
9
  annotations:
10
    strimzi.io/use-connector-resources: "true"
11
spec:
12
  replicas: 1
13
  image: "quay.io/edeandrea/kafka-connect-debezium-mysql:amq-streams-1.4.0-dbz-1.1.0.Final"
14
  bootstrapServers: "db-events-kafka-bootstrap:9092"
15
  jvmOptions:
16
    gcLoggingEnabled: false
17
  config:
18
    group.id: spring-music-db
19
    offset.storage.topic: spring-music-db-offsets
20
    config.storage.topic: spring-music-db-configs
21
    status.storage.topic: spring-music-db-status
22
    config.storage.replication.factor: 1
23
    offset.storage.replication.factor: 1
24
    status.storage.replication.factor: 1
25
    config.providers: file
26
    config.providers.file.class: org.apache.kafka.common.config.provider.FileConfigProvider
27
  externalConfiguration:
28
    volumes:
29
      - name: connector-config
30
        secret:
31
          secretName: db-connector-creds
32
  template:
33
    deployment:
34
      metadata:
35
        labels:
36
          app: spring-music-cdc
37
          app.kubernetes.io/part-of: spring-music-cdc
38
          template: spring-music-cdc
39
        annotations:
40
          app.openshift.io/connects-to: db-events-kafka,spring-music-db


This action will deploy a one-node KafkaConnect cluster. It will also turn down the JVM's garbage collection logging so that if we need to look at the logs in any of the KafkaConnect pods, they won’t be polluted with tons of garbage collection debug logs.

As you can see from this configuration, we use the quay.io/edeandrea/kafka-connect-debezium-mysql:amq-streams-1.4.0-dbz-1.1.0.Final image on line 13:

Java
 




xxxxxxxxxx
1


 
1
image: "quay.io/edeandrea/kafka-connect-debezium-mysql:amq-streams-1.4.0-dbz-1.1.0.Final"


Then we tell the KafkaConnect cluster to connect to the db-events-kafka-bootstrap:9092 bootstrap server on line 14:

Java
 




xxxxxxxxxx
1


 
1
bootstrapServers: "db-events-kafka-bootstrap:9092"


We’ve also added some externalConfiguration, which tells the KafkaConnect container to mount the secret named db-connector-creds into the directory /opt/kafka/external-configuration/connector-config within the running container (lines 27-31):

Java
 




xxxxxxxxxx
1


 
1
externalConfiguration:
2
  volumes:
3
    - name: connector-config
4
      secret:
5
        secretName: db-connector-creds


If you go back to the OpenShift Developer perspective’s Topology view, you should now see the db-events-connect deployment with one replica available, as shown in Figure 17. It might take a few minutes for it to become available.

Wait for Kafka Connect to become available

Figure 17: Wait for Kafka Connect to become available


Deploy the Debezium Connector

Now that our Kafka Connect cluster is up and running we can deploy our Debezium connector configuration into it. Back in the OpenShift console, go back to the Administrator perspective, then Installed Operators, and then click the Red Hat Integration - AMQ Streams operator, as shown in Figure 18.

Installed Operators

Figure 18: Installed Operators


Then under Provided APIs, click the Create Instance label in the Kafka Connector section, as shown in Figure 19.

Create Kafka Connector instance

Figure 19: Create Kafka Connector instance


The Create KafkaConnector YAML editor will then come up. Remove everything that's there, paste in the following, and click the Create button at the bottom of the screen. This action will deploy the connector configuration into the Kafka Connect cluster and start the connector:

Java
 




xxxxxxxxxx
1
36


 
1
kind: KafkaConnector
2
apiVersion: kafka.strimzi.io/v1alpha1
3
metadata:
4
  name: db-events
5
  namespace: debezium-demo
6
  labels:
7
    app: spring-music-cdc
8
    strimzi.io/cluster: db-events
9
spec:
10
  class: io.debezium.connector.mysql.MySqlConnector
11
  tasksMax: 1
12
  config:
13
    database.hostname: spring-music-db
14
    database.port: 3306
15
    database.user: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbUsername}"
16
    database.password: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbPassword}"
17
    database.dbname: music
18
    database.server.name: spring-music
19
    database.server.id: 223344
20
    database.whitelist: music
21
    database.allowPublicKeyRetrieval: true
22
    database.history.kafka.bootstrap.servers: db-events-kafka-bootstrap:9092
23
    database.history.kafka.topic: dbhistory.music
24
    table.whitelist: music.outbox_events
25
    tombstones.on.delete : false
26
    transforms: outbox
27
    transforms.outbox.type: io.debezium.transforms.outbox.EventRouter
28
    transforms.outbox.route.topic.replacement: "outbox.${routedByValue}.events"
29
    transforms.outbox.table.field.event.id: event_id
30
    transforms.outbox.table.field.event.key: aggregate_id
31
    transforms.outbox.table.field.event.timestamp: event_timestamp
32
    transforms.outbox.table.field.event.type: event_type
33
    transforms.outbox.table.field.event.payload.id: aggregate_id
34
    transforms.outbox.route.by.field: aggregate_type
35
    transforms.outbox.table.fields.additional.placement: "event_id:envelope:eventId,event_timestamp:envelope:eventTimestamp,aggregate_id:envelope:aggregateId,aggregate_type:envelope:aggregateType"
36
 
          


This configuration provides lots of information. You’ll notice that the database username and password are injected into the configuration via the connector.properties file stored in our OpenShift Secret on lines 15-16:

Java
 




xxxxxxxxxx
1


 
1
database.user: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbUsername}"
2
database.password: "${file:/opt/kafka/external-configuration/connector-config/connector.properties:dbPassword}"


The configuration also instructs Debezium as to which topic to place events on (line 28):

Java
 




xxxxxxxxxx
1


 
1
transforms.outbox.route.topic.replacement: "outbox.${routedByValue}.events"


Debezium supports placing all events on a single topic or using a derived routing key to decide the topic. In our case, our application only deals with a single type of domain for its events. For our application, all of the events are stored in the outbox.Album.events topic.

Note: If our application worked with more than one kind of domain event that might be unrelated to another, it might make sense to place each domain’s events into different topics.

Debezium provides a single message transformation to provide out-of-the-box support for applications implementing the Outbox pattern. More documentation on the specifics of Debezium’s Outbox Event Router and it’s configuration can be found in the Debezium documentation. Since the connector has this capability built-in, we just need to tell Debezium how to map between the fields it expects in the payload and the fields in our actual database table (lines 29-35):

Java
 




xxxxxxxxxx
1


 
1
transforms.outbox.table.field.event.id: event_id
2
transforms.outbox.table.field.event.key: aggregate_id
3
transforms.outbox.table.field.event.timestamp: event_timestamp
4
transforms.outbox.table.field.event.type: event_type
5
transforms.outbox.table.field.event.payload.id: aggregate_id
6
transforms.outbox.route.by.field: aggregate_type
7
transforms.outbox.table.fields.additional.placement: "event_id:envelope:eventId,event_timestamp:envelope:eventTimestamp,aggregate_id:envelope:aggregateId,aggregate_type:envelope:aggregateType"


We could have named the fields in our table exactly as the Debezium EventRouter transformation was looking for it, but that would then have tightly-coupled our database schema to Debezium. As a best practice, we want our components to be loosely-coupled and updateable via external configuration.

Now, how do we know this all worked? We can go directly to one of the Kafka broker pods and run the kafka-console-consumer utility to see the data in the topic.

Look at Resulting Events

Go back to the OpenShift web console and the Topology view. Click the db-events-kafka resource. When the sidebar appears on the right, click any of the three db-events-kafka pods that show up (i.e., the list in Figure 20). It doesn’t matter which one.

Select Kafka broker pod

Figure 20: Select Kafka broker pod


From there, click the Terminal tab to bring you to the terminal. Once at the terminal, run:

Java
 




xxxxxxxxxx
1


 
1
$ bin/kafka-console-consumer.sh --bootstrap-server db-events-kafka-bootstrap:9092 --topic outbox.Album.events --from-beginning


It will output a bunch of JSON, as shown in Figure 21.

Run kafka-console-consumer.sh

Figure 21: Run kafka-console-consumer.sh


You can now examine the raw output. It should look something like this:

Java
 




xxxxxxxxxx
1


 
1
{"schema":{"type":"struct","fields":[{"type":"string","optional":true,"name":"io.debezium.data.Json","version":1,"field":"payload"},{"type":"string","optional":false,"field":"eventType"},{"type":"int64","optional":false,"field":"eventId"},{"type":"int64","optional":false,"name":"io.debezium.time.MicroTimestamp","version":1,"field":"eventTimestamp"},{"type":"string","optional":false,"field":"aggregateId"},{"type":"string","optional":false,"field":"aggregateType"}],"optional":false},"payload":{"payload":"{\"album\": {\"id\": \"9d0a7606-d933-4026-9f33-efa2bde4b9e4\", \"genre\": \"Rock\", \"title\": \"Nevermind\", \"artist\": \"Nirvana\", \"albumId\": null, \"trackCount\": 0, \"releaseYear\": \"1991\"}, \"eventType\": \"ALBUM_CREATED\"}","eventType":"ALBUM_CREATED","eventId":1,"eventTimestamp":1586264029784000,"aggregateId":"9d0a7606-d933-4026-9f33-efa2bde4b9e4","aggregateType":"Album"}}


The output might not look too legible, but if you pretty-print it (Google json pretty print in your browser and find a free utility) you’ll see that the payload format looks like this:

JSON
 




xxxxxxxxxx
1
50


 
1
{
2
  "schema": {
3
    "type": "struct",
4
    "fields": [
5
      {
6
        "type": "string",
7
        "optional": true,
8
        "name": "io.debezium.data.Json",
9
        "version": 1,
10
        "field": "payload"
11
      },
12
      {
13
        "type": "string",
14
        "optional": false,
15
        "field": "eventType"
16
      },
17
      {
18
        "type": "int64",
19
        "optional": false,
20
        "field": "eventId"
21
      },
22
      {
23
        "type": "int64",
24
        "optional": false,
25
        "name": "io.debezium.time.MicroTimestamp",
26
        "version": 1,
27
        "field": "eventTimestamp"
28
      },
29
      {
30
        "type": "string",
31
        "optional": false,
32
        "field": "aggregateId"
33
      },
34
      {
35
        "type": "string",
36
        "optional": false,
37
        "field": "aggregateType"
38
      }
39
    ],
40
    "optional": false
41
  },
42
  "payload": {
43
    "payload": "{\"album\": {\"id\": \"9d0a7606-d933-4026-9f33-efa2bde4b9e4\", \"genre\": \"Rock\", \"title\": \"Nevermind\", \"artist\": \"Nirvana\", \"albumId\": null, \"trackCount\": 0, \"releaseYear\": \"1991\"}, \"eventType\": \"ALBUM_CREATED\"}",
44
    "eventType": "ALBUM_CREATED",
45
    "eventId": 1,
46
    "eventTimestamp": 1586264029784000,
47
    "aggregateId": "9d0a7606-d933-4026-9f33-efa2bde4b9e4",
48
    "aggregateType": "Album"
49
  }
50
}


This payload defines its own structure in the schema element on lines 2-41:

Java
 




xxxxxxxxxx
1
20


 
1
"schema": { 
2
  "type": "struct", 
3
  "fields": [ 
4
    { 
5
      "type": "string", 
6
      "optional": true, 
7
      "name": "io.debezium.data.Json", 
8
      "version": 1, 
9
      "field": "payload" 
10
    },
11
    { 
12
      "type": "string", 
13
      "optional": false, 
14
      "field": "eventType"
15
    },
16
    {
17
      "type": "int64",
18
      "optional": false,
19
      "field": "eventId"
20
    }, { "type": "int64", "optional": false, "name": "io.debezium.time.MicroTimestamp", "version": 1, "field": "eventTimestamp" }, { "type": "string", "optional": false, "field": "aggregateId" }, { "type": "string", "optional": false, "field": "aggregateType" } ], "optional": false }


We could eliminate this section by standing up our own schema registry and configuring our cluster to use Avro serialization/deserialization. The payload element on lines 42-49 contains metadata about the event, as well as the actual payload of the event:

Java
 




xxxxxxxxxx
1


 
1
"payload": { 
2
  "payload": "{\"album\": {\"id\": \"9d0a7606-d933-4026-9f33-efa2bde4b9e4\", \"genre\": \"Rock\", \"title\": \"Nevermind\", \"artist\": \"Nirvana\", \"albumId\": null, \"trackCount\": 0, \"releaseYear\": \"1991\"}, \"eventType\": \"ALBUM_CREATED\"}", 
3
  "eventType": "ALBUM_CREATED", 
4
  "eventId": 1, 
5
  "eventTimestamp": 1586264029784000, 
6
  "aggregateId": "9d0a7606-d933-4026-9f33-efa2bde4b9e4", 
7
  "aggregateType": "Album" 
8
}


The payload sub-element within the main payload element (line 43) is itself a JSON string representing the contents of the domain object making up the event:

Java
 




xxxxxxxxxx
1


 
1
"payload": "{\"album\": {\"id\": \"9d0a7606-d933-4026-9f33-efa2bde4b9e4\", \"genre\": \"Rock\", \"title\": \"Nevermind\", \"artist\": \"Nirvana\", \"albumId\": null, \"trackCount\": 0, \"releaseYear\": \"1991\"}, \"eventType\": \"ALBUM_CREATED\"}"


If you keep this terminal window open and open up a new browser window back to the application itself, you should see new events stream in as you update/delete albums from the application’s user interface.

Next Steps

Hopefully, you found this post helpful! If so, please watch for a few other posts in this series once they become available:

  • Adding Prometheus metrics and Grafana Dashboard monitoring
  • Securing Kafka and KafkaConnect with OAuth authentication
  • Adding access control to Kafka and KafkaConnect with OAuth authorization

Also, if you are like me and want to automate the provisioning of everything, feel free to take a look at an Ansible Playbook that is capable of doing this.

References

This post was originally published on Red Hat Developer. To read the original post, visit https://developers.redhat.com/blog/2020/05/08/change-data-capture-with-debezium-a-simple-how-to-part-1/.

Topics:
change data capture ,debezium ,event-driven ,integration ,java ,kafka ,kubernetes ,microservices ,spring boot ,tutorial

Published at DZone with permission of Eric Deandrea . See the original article here.

Opinions expressed by DZone contributors are their own.

{{ parent.title || parent.header.title}}

{{ parent.tldr }}

{{ parent.urlSource.name }}