Getting Started with Debezium on OpenShift

Difficulty: intermediate

Estimated Time: 30 minutes

What you will learn

In this scenario you will learn more about Debezium, a project that provides change data capture for any of supported databases

MySQL
PostgreSQL
MongoDB
Microsoft SQL Server
Oracle (incubating)
Apache Cassandra (incubating)

You will deploy a complete end-to-end solution that will capture events from database transaction logs and make those events available to processing by downstream consumers via an Apache Kafka broker.

What is Debezium?

Debezium is a set of distributed services capture row-level changes in your databases so that your applications can see and respond to those changes. Debezium records all row-level changes committed to a particular database table in a dedicated message topic. Each application simply reads the topic(s) they are interested in, and they see all of the events in the same order in which they occurred.

Technically Debezium utilizes the Apache Kafka streaming platform to distribute events captured from database. It is a set of plug-ins for Kafka Connect that publish messages to a Kafka broker.

The minimum components required for skeleton deployment are

Kafka broker - consisting of a single Apache ZooKeeper instance for cluster management and a single node of Kafka broker
Kafka Connect node - containing and configured to stream data from a database
source database

The following diagram shows the minimal deployment

Minimal deployment

In the next steps we will deploy the components and get dataflow running from a MySQL database to a Kafka broker.

In this scenario you learned about the change data capture concept and how you can leverage Debezium for that purpose.

You have learnt what components you need to deploy a solution based on Debezium, how to deploy an Apache Kafka broker and how to deploy a Kafka Connect instance with Debezium inside and create a link between the Kafka Connect and source database.

But this is just a beginning of a long journey. Please take your time and look at these resources:

Debezium website
Extensive Debezium tutorial
Debezium examples repository
How to run Debezium on Minishift
The Strimzi project for running Apache Kafka on OpenShift

Your environment is currently being packaged as a Docker container and the download will begin shortly. To run the image locally, once Docker has been installed, use the commands

cat scrapbook_openshift/middleware_debezium-getting-started_container.tar | docker load

docker run -it /openshift/middleware_debezium-getting-started:20210924

Restart Scenario

Next Scenario

Step 1 of 3

Deploying a Kafka broker

A fresh project named debezium is prepared with the necessary resources required to execute the deployment. There are multiple resources created for you in the home directory, the project itself or configured in OpenShift

an installed release 0.14.0 of Strimzi project Kafka operator
Strimzi Cluster Operator managing Kafka brokers
MySQL instance containing a small set of data to be streamed
templates used to deploy components

1. Run the following commands to switch to debezium project and explore it.

If you click on command it gets automatically copied it into the terminal and is executed

Switch to debezium project

oc project debezium

Check that MySQL instance is running

oc get pods

and that it is exposed as a service

oc get svc

The diagram of deployment now looks like

Empty deployment

2. Deploy Kafka broker with ZooKeeper.

The first component to deploy is a Kafka broker.

Broker deployment

This task is delegated to templates and Cluster Controller provided by Strimzi project. The templates are already present in the home directory in the cloned repository.

The templates by default deploy Kafka broker and ZooKeeper in a high-available configuration with replication factor 3. This is not necessary in the development environment so we reduce the number of nodes and replication factor for system topics to 1.

We also deploy an ephemeral variant of the broker. You should use persistent variant in production.

To deploy the broker issue a command

oc new-app strimzi-ephemeral -p ZOOKEEPER_NODE_COUNT=1 -p KAFKA_NODE_COUNT=1 -p KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR=1 -p KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR=1

Now let's wait till both ZooKeeper and Kafka broker are deployed

oc get pods -w

The final list of pods should be similar to

NAME                                          READY     STATUS    RESTARTS   AGE
my-cluster-entity-operator-798b74565c-bkjwh   3/3       Running   1          32s
my-cluster-kafka-0                            2/2       Running   0          1m
my-cluster-zookeeper-0                        2/2       Running   0          1m
mysql-1-w7shk                                 1/1       Running   0          9m
strimzi-cluster-operator-5658b55c84-89mf5     1/1       Running   0          9m

Note: Kafka depends on ZooKeeper so intermittent Kafka failures are expected as ZooKeeper might not be initialized at the time of Kafka start.

New services are available

oc get svc -l app=strimzi-ephemeral

NAME                          TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)                      AGE
my-cluster-kafka-bootstrap    ClusterIP   172.30.136.36   <none>        9091/TCP,9092/TCP,9093/TCP   2m
my-cluster-kafka-brokers      ClusterIP   None            <none>        9091/TCP,9092/TCP,9093/TCP   2m
my-cluster-zookeeper-client   ClusterIP   172.30.82.207   <none>        2181/TCP                     3m
my-cluster-zookeeper-nodes    ClusterIP   None            <none>        2181/TCP,2888/TCP,3888/TCP   3m

3. Verify the broker is up and running.

Note: The complete initialization of all components can take a couple of minutes. Please make sure that all pods are in Running state and are Ready before you try the next steps.

A successful attempt to send a message to (no output expected here)

echo "Hello world" | oc exec -i -c kafka my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

and receive a message from

oc exec -c kafka my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic test --from-beginning --max-messages 1

the deployed broker indicates that it is available.

Congratulations

You have now successfully executed the first step in this scenario.

You have successfully deployed Kafka broker service and made it available to clients to produce and consume messages.

In the next step of this scenario, we will deploy a single instance of Debezium.

Deploying a Kafka Connect with Debezium

We are going to use binary build feature of OpenShift together with distribution of MySQL plugin to deploy a Kafka Connect node with Debezium MySQL plugin embedded.

Debezium deployment

1. Deploy an empty Kafka Connect node

We will again use a template from Strimzi project that is already installed in the environment. As with Kafka broker we need to configure the template to support single-instance-only deployment.

oc new-app strimzi-connect-s2i -p KAFKA_CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR=1 -p KAFKA_CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR=1 -p KAFKA_CONNECT_STATUS_STORAGE_REPLICATION_FACTOR=1

A new build config will be created

oc get bc

NAME                         TYPE      FROM      LATEST
my-connect-cluster-connect   Source    Binary    1

and a new service is deployed.

oc get svc -l app=strimzi-connect-s2i

NAME                             TYPE        CLUSTER-IP      EXTERNAL-IP   PORT(S)    AGE
my-connect-cluster-connect-api   ClusterIP   172.30.89.106   <none>        8083/TCP   11s

2. Embed Debezium into Connect

We will use a binary build to create a Connect node with a Debezium plugin inside

oc start-build my-connect-cluster-connect --from-archive https://repo.maven.apache.org/maven2/io/debezium/debezium-connector-mysql/1.1.1.Final/debezium-connector-mysql-1.1.1.Final-plugin.tar.gz --follow

The Connect node should be redeployed after the build completes

oc get pods -w -l app=strimzi-connect-s2i

NAME                                 READY     STATUS        RESTARTS   AGE
my-connect-cluster-connect-1-wktnt   0/1       Terminating   0          50s
my-connect-cluster-connect-2-lqqht   0/1       Running       0          26s

3. Verify that Connect is up and contains Debezium

Wait till the Connect node is ready

NAME                                 READY     STATUS    RESTARTS   AGE
my-connect-cluster-connect-2-lpnd2   1/1       Running   0          1m

and list all plug-ins available for use.

oc exec -c kafka my-cluster-kafka-0 -- curl -s http://my-connect-cluster-connect-api:8083/connector-plugins

[
    {"class":"io.debezium.connector.mysql.MySqlConnector","type":"source","version":"0.10.0.Final"},
    {"class":"org.apache.kafka.connect.file.FileStreamSinkConnector","type":"sink","version":"2.3.0"},
    {"class":"org.apache.kafka.connect.file.FileStreamSourceConnector","type":"source","version":"2.3.0"}
]

The Connect node has the Debezium MySqlConnector connector plugin available.

Congratulations

You have now successfully executed the second step in this scenario.

You have successfully deployed Kafka Connect node and configured it to contain Debezium.

In next step of this scenario we will finish deployment by creating a connection between database source and Kafka Connect.

Streaming data from database

The last step to do is to create a link between Debezium and a source MySQL database. This is achieved via Kafka Connect REST API configuration.

Complete deployment

1. Register database source

To register the source we need to send a POST request to the /connectors endpoint of Kafka Connect API. An example registration request is part of the evironment - register.json.

To register the database source execute

cat register.json | oc exec -i -c kafka my-cluster-kafka-0 -- curl -s -X POST -H "Accept:application/json" -H "Content-Type:application/json" http://my-connect-cluster-connect-api:8083/connectors -d @-

Check the Connect's log file to see that the registration has succeeded and change data capture has started

oc logs $(oc get pods -o name -l app=strimzi-connect-s2i)

Now Kafka topics are created when the connector starts to capture database changes. Those topics could be listed using command

oc exec -c kafka my-cluster-kafka-0 -- /opt/kafka/bin/kafka-topics.sh --bootstrap-server my-cluster-kafka-bootstrap:9092 --list

And the topics corresponding to the database tables are

dbserver1.inventory.addresses
dbserver1.inventory.customers
dbserver1.inventory.geom
dbserver1.inventory.orders
dbserver1.inventory.products
dbserver1.inventory.products_on_hand

2. Verify that data are sourced from the MySQL server to the Kafka broker

Let's check the contents of customers table in MySQL. The command

oc exec -i $(oc get pods -o custom-columns=NAME:.metadata.name --no-headers -l app=mysql) -- bash -c 'mysql -t -u $MYSQL_USER -p$MYSQL_PASSWORD -e "SELECT * from customers" inventory'

should yield a result

+------+------------+-----------+-----------------------+
| id   | first_name | last_name | email                 |
+------+------------+-----------+-----------------------+
| 1001 | Sally      | Thomas    | [email protected] |
| 1002 | George     | Bailey    | [email protected]    |
| 1003 | Edward     | Walker    | [email protected]         |
| 1004 | Anne       | Kretchmar | [email protected]    |
+------+------------+-----------+-----------------------+

The Kafka broker should contain an equivalent list of massages in topic dbserver1.inventory.customers in the Debezium change event format

oc exec -c kafka my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dbserver1.inventory.customers --from-beginning --max-messages 4

Note: Output formatted for the sake of readability

...
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_sec"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "boolean",
            "optional": true,
            "field": "snapshot"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "dbserver1.inventory.customers.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "id": 1004,
      "first_name": "Anne",
      "last_name": "Kretchmar",
      "email": "[email protected]"
    },
    "source": {
      "name": "dbserver1",
      "server_id": 0,
      "ts_sec": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 154,
      "row": 0,
      "snapshot": true,
      "thread": null,
      "db": "inventory",
      "table": "customers"
    },
    "op": "c",
    "ts_ms": 1513066125526
  }
}
Processed a total of 4 messages

If we add a new record to the table

oc exec -i $(oc get pods -o custom-columns=NAME:.metadata.name --no-headers -l app=mysql) -- bash -c 'mysql -t -u $MYSQL_USER -p$MYSQL_PASSWORD -e "INSERT INTO customers VALUES(default,\"John\",\"Doe\",\"[email protected]\")" inventory'

a new message will be sent to the associated topic

oc exec -c kafka my-cluster-kafka-0 -- /opt/kafka/bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --topic dbserver1.inventory.customers --from-beginning --max-messages 5

...
{
  "schema": {
    "type": "struct",
    "fields": [
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "before"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "int32",
            "optional": false,
            "field": "id"
          },
          {
            "type": "string",
            "optional": false,
            "field": "first_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "last_name"
          },
          {
            "type": "string",
            "optional": false,
            "field": "email"
          }
        ],
        "optional": true,
        "name": "dbserver1.inventory.customers.Value",
        "field": "after"
      },
      {
        "type": "struct",
        "fields": [
          {
            "type": "string",
            "optional": false,
            "field": "name"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "server_id"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "ts_sec"
          },
          {
            "type": "string",
            "optional": true,
            "field": "gtid"
          },
          {
            "type": "string",
            "optional": false,
            "field": "file"
          },
          {
            "type": "int64",
            "optional": false,
            "field": "pos"
          },
          {
            "type": "int32",
            "optional": false,
            "field": "row"
          },
          {
            "type": "boolean",
            "optional": true,
            "field": "snapshot"
          },
          {
            "type": "int64",
            "optional": true,
            "field": "thread"
          },
          {
            "type": "string",
            "optional": true,
            "field": "db"
          },
          {
            "type": "string",
            "optional": true,
            "field": "table"
          }
        ],
        "optional": false,
        "name": "io.debezium.connector.mysql.Source",
        "field": "source"
      },
      {
        "type": "string",
        "optional": false,
        "field": "op"
      },
      {
        "type": "int64",
        "optional": true,
        "field": "ts_ms"
      }
    ],
    "optional": false,
    "name": "dbserver1.inventory.customers.Envelope",
    "version": 1
  },
  "payload": {
    "before": null,
    "after": {
      "id": 1005,
      "first_name": "John",
      "last_name": "Doe",
      "email": "[email protected]"
    },
    "source": {
      "name": "dbserver1",
      "server_id": 0,
      "ts_sec": 0,
      "gtid": null,
      "file": "mysql-bin.000003",
      "pos": 465,
      "row": 0,
      "snapshot": true,
      "thread": null,
      "db": "inventory",
      "table": "customers"
    },
    "op": "c",
    "ts_ms": 1513068550815
  }
}
Processed a total of 5 messages

Congratulations

You have completed the basic deployment scenario. Now you can add or remove data from MySQL and see the results in Kafka Broker.

See Debezium's extensive tutorial for more tips and details.

Terminal

Dashboard

cd <directory>	Change directory
ls	List directory
echo 'contents' > <file>	Write contents to a file
cat <file>	Output contents of file

i	In Command Mode	Change into Insert Mode, you can now insert and edit text in the file
esc	In Insert Mode	Change into Command Mode, you can now execute commands
:wq	In Command Mode	Save and Exit

Welcome!

Getting Started with Debezium on OpenShift

What you will learn

What is Debezium?

Congratulations!

You've completed the scenario!

Scenario Rating

Steps