Change Data Capture from Amazon RDS to Apache Kafka® with Debezium
Tags: cdc, apachekafka, rds, debezium
Published: 2023-10-17 16:00:00

---
title: Change Data Capture from Amazon RDS to Apache Kafka® with Debezium
published: true
date: 2023-10-17 16:00 UTC
tags: cdc, apachekafka, rds, debezium
canonical_url: https://aiven.io/developer/cdc-amazon-rds-to-apache-kafka-debezium
---

[Amazon RDS](https://aws.amazon.com/rds/) offers a PostgreSQL backend for applications. This tutorial show you how to build a real-time change data capture process to track the changes happening in one (or more) RDS tables and stream them into Apache Kafka®, where multiple consumers can receive the data with a minimal latency.

We'll use [Aiven for Apache Kafka®](https://aiven.io/kafka?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign) as the destination for our streaming data. 

## Create AWS RDS database

Head to the [AWS Console](https://aws.amazon.com/console/) and:

* Navigate to the **Products** tab
* Select **Databases**
* Click on **RDS**
* Click on **Create Database**

![Create database button](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/c0vhwj4px0dfxkrg8vut.png)

* Select **PostgreSQL**
* Give the database a name like `database-test`
* In the **Availability and durability** section, select **Single DB instance** (for the purpose of this tutorial, the Multi-AZ DB instance and Multi-AZ DB Cluster are functionally equivalent)
* In the **Settings** section, change the master password to `test12345`

Note: For production use cases, use a secure password.

* In the **Connectivity** section, select a VPC with an [internet gateway attached](https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_VPC.WorkingWithRDSInstanceinaVPC.html) and enable **Public Access**. You also might want to check that the inbound traffic is allowed from the IP you'll try to connect to RDS from.

Note: you can avoid exposing RDS to public access and connect it to Apache Kafka via [VPC peering](https://docs.aiven.io/docs/platform/howto/manage-vpc-peering) or have the [Aiven for Apache Kafka®](https://aiven.io/kafka?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign) service as part of your AWS account with the [Bring Your Own Cloud model](https://docs.aiven.io/docs/platform/concepts/byoc?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign).

## Populate AWS RDS database

Once the database is created, we can review the details in the AWS console. In the **Connectivity and Security** section we can check the database hostname and port.

![AWS Console showing the database details](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/bhnh9m81lb26f6hanfhv.png)

With the above information, we can connect to RDS with [psql](https://docs.aiven.io/docs/products/postgresql/howto/connect-psql?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign):

```bash
psql postgres://<USERNAME>:<PASSWORD>@<HOSTNAME>:<PORT>/<DATABASE_NAME>
```

Where:

* `<USERNAME>` is the connection username, `postgres` if you left the default unchanged
* `<PASSWORD>` is the connection password, `test12345` if you followed the above instructions
* `<HOSTNAME>` is the database host, taken from the AWS Console
* `<PORT>` is the database port, taken from the AWS Console
* `<DATABASE_NAME>` is the database name, by default `postgres`

If you're using the defaults and have followed the above instructions, the `psql` call should be:

```bash
psql postgres://postgres:test12345@<HOSTNAME>:<PORT>/postgres
```

Once connected, we can create a test table and insert some data:

```sql
CREATE TABLE FORNITURE (ID SERIAL, NAME TEXT);

INSERT INTO FORNITURE (NAME) VALUES ('CHAIR'),('TABLE'),('SOFA'),('FRIDGE');
```

A query like `SELECT * FROM FORNITURE;` should provide the following results

```bash
 id |  name
----+--------
  1 | CHAIR
  2 | TABLE
  3 | SOFA
  4 | FRIDGE
(4 rows)
```

## Create an Aiven for Apache Kafka® service with Kafka Connect enabled

If you don't have an Apache Kafka cluster available, you can create one with Aiven by:

* Navigate to the [Aiven Console](aiven.io/signup)
* Click on **Create service**
* Select **Apache Kafka®**
* Select the cloud and region where the service will be deployed. Selecting the same cloud region where your RDS database is located will minimize latency.
* Access the Aiven for Apache Kafka service page and enable: 
    * **Kafka Connect** to perform the change data capture
    * **REST API** to browse the data from the Aiven Console
    * ``kafka.auto_create_topics_enable`` from the **Advanced configuration** section to automatically create topics based on the Kafka Connect configurations

![Apache Kafka REST APIs and Kafka Connect enabled](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/ygqr5cojly4x5i4dcamu.png)

## Create a Change Data Capture process with the Debezium Connector

Once Aiven for Apache Kafka is running, the next step is to setup the CDC pipeline. To do so you can head to the **Connectors** tab, select the **Debezium for PostgreSQL** connector and include the following configuration:

```
{
    "name": "mysourcedebezium",
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "<HOSTNAME>",
    "database.port": "<PORT>",
    "database.user": "postgres",
    "database.password": "<PASSWORD>",
    "database.dbname": "postgres",
    "database.server.name": "mydebprefix",
    "plugin.name": "pgoutput",
    "slot.name": "mydeb_slot",
    "publication.name": "mydeb_pub",
    "publication.autocreate.mode": "filtered",
    "table.include.list": "public.forniture"
}
```

Where:

* `database.hostname`, `database.port`,  `database.password` are the RDS connection parameters found in the AWS Console
* `database.server.name` is the prefix for the topic names in Aiven for Apache Kafka
* `plugin.name` is the PostgreSQL plugin name, `pgoutput`
* `slot.name` and `publication.name` are the name of the replication slot and publication in PostgreSQL
* `"publication.autocreate.mode": "filtered"` creates a publication only for the tables in scope
* `table.include.list` lists the tables for which we want to enable CDC

After replacing the placeholders in the JSON configuration file with the connection parameters defined above you can start the connector by:

* Navigating to the [Aiven Console](https://console.aiven.io/?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign)
* Navigating to the Aiven for Apache Kafka service page
* Clicking in the **Connectors** tab
* Clicking on **New Connector**
* Selecting the **Debezium - PostgreSQL**
* Editing the JSON connector configuration and pasting the JSON configuration defined above.
* Clicking on **Create Connector**


If you hit the error below

```
There was an error in the configuration.
database.hostname: Postgres server wal_level property must be "logical" but is: replica
```

You'll need to enable [logical replication](https://docs.aiven.io/docs/products/postgresql/howto/logical-replication-aws-rds?utm_source=devto&utm_medium=referral&utm_campaign=dev-one-api-campaign). To check the logical replication run `show wal_level;` from a terminal connected to the PostgreSQL database, it should show the `wal_level` as `logical`.

## Check the changes in Apache Kafka

Once the connector is up and running, you should see a topic named ``mydebprefix.public.forniture``, the concatenation of the ``database.server.name`` parameter and the RDS schema and table name. To check the data in the topic in the Aiven Console:

* Navigate to the **Topics** tab
* Click on the ``mydebprefix.public.forniture`` topic
* Click on **Messages**
* Click on **Fetch Messages**
* Enable the **Decode from base64**

You should see the same dataset you previously pushed to RDS appearing in JSON format in the topic.

![CDC data appearing in Apache Kafka](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/l0vsjcpwjw0gfkohp6pq.png)

If we perform an insert, delete and update using psql in the terminal connected to RDS, for example:

```sql
INSERT INTO FORNITURE (NAME) VALUE ('REFRIGERATOR');
DELETE FROM FORNITURE WHERE NAME='FRIDGE';
UPDATE FORNITURE SET NAME='COUCH' WHERE NAME='SOFA';
```

You might get the following error:

```bash
ERROR:  cannot delete from table "forniture" because it does not have a replica identity and publishes deletes
HINT:  To enable deleting from the table, set REPLICA IDENTITY using ALTER TABLE.
```

To solve the problem you can enable [full replica identity](https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY) in the table with:

```sql
ALTER TABLE FORNUTURE SET REPLICA IDENTITY FULL;
```

If the above changes work, we should be able to see them in the Aiven for Kafka UI, by re-clicking on the **Fetch Messages** button.

![The last three changes being visualized in Aiven for Apache Kafka topic](https://dev-to-uploads.s3.amazonaws.com/uploads/articles/bm5r08vk44n4j7sdxjix.png)

## Conclusion

Setting up a change data capture process from an RDS PostgreSQL database to Apache Kafka with the Debezium connector is a powerful method to be able to stream the inserts/updates/deletes to one or more consumers in real time. 

Some more resources if you are interested:

* [Debezium PostgreSQL connector parameters](https://debezium.io/documentation/reference/stable/connectors/postgresql.html)
* [PostgreSQL replica identity](https://www.postgresql.org/docs/current/sql-altertable.html#SQL-ALTERTABLE-REPLICA-IDENTITY)
* [Limits of the JDBC source connector](https://www.confluent.io/events/kafka-summit-london-2022/jdbc-source-connector-what-could-go-wrong/)