Change Data Capture —Using Debezium

This article will help you in integrating Debezium to capture the changes in the data events, as well as it could help you in identifying the way to decompose monolithic applications.

Photo by Margaret Weir on Unsplash

I am writing this article to explain the recent problem I encountered while decomposing a monolithic application to an event-driven architecture based micro-services application for one of our clients.

The application is a vastly used E-commerce product and customer based increase exponentially in the last few years.

At the start of the last year, before the lockdown situation happened, the client predicted that there would be an increase of load on the application.

And yes there was a drastic increase in customer orders, transactions, and the customer base, during the pandemic house arrest.

Furthermore, To proactively handle this, the client-initiated their most prominent technical debt that was to move towards event-driven-based microservice and to get rid of their current monolith architecture problems.

The major problem with the current architecture is that it is running at its full capacity, adding more hardware resources does not bring any positive improvement on the other hand it increased the deployment time and maintenance window.

Second, the Time to market of an average feature or change is too high as adding a single feature code change required senior developers and cross-team discussion which impacts their productivity and quality.

Last, Maintenance Cost is too high. So many developers working on the same Mono repo, improper visibility and lack of knowledge plus documentation makes new bees reluctant to jump into the battlefield.

Once I got onboarded with my team of three SDEs in to this project. We discussed with the senior developers who are working on the project for the last few years. And they helped me in discovering the loopholes in design and architecture.

We also observed their daily tasks and created a list of the immediate actionable.

• Deployment issue — It requires a dedicated team. It is quite time-consuming and only happens at night.
• Observability issue — No central logging, requires few manual jobs to run and have operation team dependencies
• Manual Mundane tasks — post-deployment as well as regular reports generation
• Performance Issue — Because of the poorly implemented custom cache, load enormous data post-application restart.

Performance issue

We set up the dedicated Redis cache server which improves the performance. For most of the use cases, application modules were directly accessing the database and third party systems repeatedly.

As Redis can save data on disk that makes the application more fault tolerant. As it prevents the downstream applications or database to get overwhelmed requests in case the server get crashed.

No doubt, Cache improves overall performance by reducing the database and network hits for the read-only data that are required for a specific period of time.

Decomposition

We observed this monolith application was of Modular Monolith sub-category where most of the modules have the identical architecture. And from a data domain perspective, most of them were independent, but few of them required some information about the customers or inventory from the master database. So, we initiated the decomposition of such modules into the separately deployable micro-services and moved the common POJO as common artifacts. Only to reduce Quantum size and improve deployment time and effort.

Once we decomposed one major services into independent micro-service, it solved some of our problems related to performance and deployment. But the challenge was to get the latest customer details from monolith application.

As making code change in monolith application was not an option, because of various constraints. Another, to notify newly decomposed microservice via REST API required code change at lots of classes. We tried to use Cross-Cutting concerns but it did not help, then we focus on the last approach that was Change Data Capture(CDC), which many of you will not approve.

As CDC helps you in solving one problem but it introduce few another (I will talk about them in other article).

We implemented a CDC solution by using Debezium connector which monitor any change in the customer table and send an event into the Kafka streaming service. Then newly created Microservice updating its table by consuming change data events.

Here, I will be explaining how to do implement CDC via Debezium. Refer below repository for the source code.

https://github.com/RitreshGirdhar/Change-data-capture-demo

We require below components to be running. For this demo, I am using docker.

Run Mysql

docker run -it --rm --name mysql -p 3306:3306 -e MYSQL_ROOT_PASSWORD=  -e MYSQL_USER=mysqluser -e MYSQL_PASSWORD=mysqlpw debezium/example-mysql:1.4

Run zookeeper

docker run -it --rm --name zookeeper -p 2181:2181 -p 2888:2888 -p 3888:3888 debezium/zookeeper:1.4

Run kafka

docker run -it --rm --name kafka -p 9092:9092 --link zookeeper:zookeeper debezium/kafka:1.4

Run kafka-connect

docker run -it --rm --name connect -p 8083:8083 -e GROUP_ID=1 -e CONFIG_STORAGE_TOPIC=my_connect_configs -e OFFSET_STORAGE_TOPIC=my_connect_offsets -e STATUS_STORAGE_TOPIC=my_connect_statuses --link zookeeper:zookeeper --link kafka:kafka --link mysql:mysql debezium/connect:1.4

Check Kafka connect service

curl -H "Accept:application/json" localhost:8083/
{"version":"2.6.0","commit":"cb8625948210849f"}

List kafka connect connectors

curl -H "Accept:application/json" localhost:8083/connectors/
[]

Registering connector to the database

> curl -i -X POST -H "Accept:application/json" -H "Content-Type:application/json" localhost:8083/connectors/ -d '{ "name": "inventory-connector", "config": { "connector.class": "io.debezium.connector.mysql.MySqlConnector", "tasks.max": "1", "database.hostname": "mysql", "database.port": "3306", "database.user": "debezium", "database.password": "dbz", "database.server.id": "184054", "database.server.name": "dbserver1", "database.include.list": "inventory", "database.history.kafka.bootstrap.servers": "kafka:9092", "database.history.kafka.topic": "dbhistory.inventory" } }'

Verify Connectors

> curl -i -X GET -H "Accept:application/json" localhost:8083/connectors/inventory-connector
HTTP/1.1 200 OK
Date: Wed, 27 Jan 2021 16:02:43 GMT
Content-Type: application/json
Content-Length: 535
Server: Jetty(9.4.33.v20201020)
{"name":"inventory-connector","config":{"connector.class":"io.debezium.connector.mysql.MySqlConnector","database.user":"root","database.server.id":"184054","tasks.max":"1","database.hostname":"mysql","database.password":"debezium","database.history.kafka.bootstrap.servers":"kafka:9092","database.history.kafka.topic":"dbhistory.inventory","name":"inventory-connector","database.server.name":"dbserver1","database.port":"3306","database.include.list":"inventory"},"tasks":[{"connector":"inventory-connector","task":0}],"type":"source"}

Watch Kafka topic

docker run -it --rm --name watcher --link zookeeper:zookeeper --link kafka:kafka debezium/kafka:1.4 watch-topic -a -k dbserver1.inventory.customers

Start Customer-Service — Here, Customer-Service microservice will make changes in the customer table in the database. Use below command to start the microservice.

> mvn clean install
> docker run -d -p 6060:8080 --link mysql:mysql customer-service

Start Invoice-Service application — Here, Invoice-Service microservice will behave as consumer which will capture the changes and update its cache or database. Use below command to start this microservice.

> mvn clean install
> docker run -d -p 7070:7070 --link mysql:mysql invoice-service

You will see below logs while changing entry in customers table records

{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1005}} {"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":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"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":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope"},"payload":{"before":null,"after":{"id":1005,"first_name":"Ritresh","last_name":"Girdhar","email":"ritresh.girdhar@gmail.com"},"source":{"version":"1.4.0.Final","connector":"mysql","name":"dbserver1","ts_ms":1611762877000,"snapshot":"false","db":"inventory","table":"customers","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":364,"row":0,"thread":5,"query":null},"op":"c","ts_ms":1611762877035,"transaction":null}}
{"schema":{"type":"struct","fields":[{"type":"int32","optional":false,"field":"id"}],"optional":false,"name":"dbserver1.inventory.customers.Key"},"payload":{"id":1005}} {"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":"version"},{"type":"string","optional":false,"field":"connector"},{"type":"string","optional":false,"field":"name"},{"type":"int64","optional":false,"field":"ts_ms"},{"type":"string","optional":true,"name":"io.debezium.data.Enum","version":1,"parameters":{"allowed":"true,last,false"},"default":"false","field":"snapshot"},{"type":"string","optional":false,"field":"db"},{"type":"string","optional":true,"field":"table"},{"type":"int64","optional":false,"field":"server_id"},{"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":"int64","optional":true,"field":"thread"},{"type":"string","optional":true,"field":"query"}],"optional":false,"name":"io.debezium.connector.mysql.Source","field":"source"},{"type":"string","optional":false,"field":"op"},{"type":"int64","optional":true,"field":"ts_ms"},{"type":"struct","fields":[{"type":"string","optional":false,"field":"id"},{"type":"int64","optional":false,"field":"total_order"},{"type":"int64","optional":false,"field":"data_collection_order"}],"optional":true,"field":"transaction"}],"optional":false,"name":"dbserver1.inventory.customers.Envelope"},"payload":{"before":{"id":1005,"first_name":"Ritresh","last_name":"Girdhar","email":"ritresh.girdhar@gmail.com"},"after":{"id":1005,"first_name":"Ritresh","last_name":"Girdhar","email":"ritresh.girdhar@yahoo.com"},"source":{"version":"1.4.0.Final","connector":"mysql","name":"dbserver1","ts_ms":1611762908000,"snapshot":"false","db":"inventory","table":"customers","server_id":223344,"gtid":null,"file":"mysql-bin.000003","pos":687,"row":0,"thread":5,"query":null},"op":"u","ts_ms":1611762908426,"transaction":null}}

Check changes in the new Database table.

Do checkout my another articles. Happy reading!

Hit 👏 button (10 times), It encourages me to keep writing.

How to effectively manage the grocery list in the era of digitalisation !? 🤔

Why to lint DockerFile in Continuous Integration?

Why to choose — Google Cloud Run & Google Scheduler for scheduling jobs ? 🤔

Alexa — Deploy my service !!!

Continuous Integration — Github-Jenkins + Telegram integration

Capture IOT sensor’s data — MQTT Protocol + RabbitMQ as MQTT Broker