Tutorial: Set up a Change Data Capture architecture on Azure using Debezium, Postgres and Kafka
Tags: azure, postgres, kafka, tutorial
Published: 2020-07-02 07:47:07

*Change Data Capture (CDC)* is a technique used to track row-level changes in database tables in response to create, update and delete operations. Different databases use different techniques to expose these change data events - for example, [logical decoding in PostgreSQL](https://www.postgresql.org/docs/current/static/logicaldecoding-explanation.html), [MySQL binary log (binlog)](https://dev.mysql.com/doc/internals/en/binary-log-overview.html) etc. This is a powerful capability, but useful only if there is a way to tap into these event logs and make it available to other services which depend on that information.

[Debezium](https://debezium.io/) does just that! It is a distributed platform that builds on top of Change Data Capture features available in different databases. It provides a set of [Kafka Connect connectors](https://debezium.io/documentation/reference/1.2/connectors/index.html) which tap into row-level changes (using CDC) in database table(s) and convert them into event streams. These event streams are sent to [Apache Kafka](https://kafka.apache.org/) which is a scalable event streaming platform - a perfect fit! Once the change log events are in Kafka, they will be available to all the downstream applications.

> This is different compared to the "polling" technique adopted by the [Kafka Connect JDBC connector](https://github.com/confluentinc/kafka-connect-jdbc)

The diagram (from the debezium.io website) summarises it nicely!

![](https://debezium.io/documentation/reference/1.2/_images/debezium-architecture.png)

This blog is a guide to getting started with setting up a change data capture based system on Azure using Debezium, [Azure DB for PostgreSQL](https://docs.microsoft.com/azure/postgresql/overview?WT.mc_id=devto-blog-abhishgu) and [Azure Event Hubs](https://docs.microsoft.com/azure/event-hubs/event-hubs-about?WT.mc_id=devto-blog-abhishgu) (for Kafka). It will use the [Debezium PostgreSQL connector](https://debezium.io/documentation/reference/1.2/connectors/postgresql.html) to stream database modifications from PostgreSQL to Kafka topics in Azure Event Hubs

> The related config files are available in the GitHub repo https://github.com/abhirockzz/

Although I have used managed Azure services for demonstration purposes these instructions should work for any other setup as well e.g. a local Kafka cluster and PostgreSQL instance.


## Setup PostgreSQL and Kafka on Azure

This section will provide pointers on how to configure Azure Event Hubs and Azure DB for PostgreSQL. All you need is a [Microsoft Azure account](https://docs.microsoft.com/azure/?WT.mc_id=devto-blog-abhishgu) - go ahead and [sign up for a free one!](https://azure.microsoft.com/free/?WT.mc_id=devto-blog-abhishgu)

### Azure DB for PostgreSQL

Azure DB for PostgreSQL is a managed, relational database service based on the community version of open-source PostgreSQL database engine, and is available in two deployment modes. 

> At the time of writing, it supports PostgreSQL version `11.6`

You can setup PostgreSQL on Azure using a variety of options including, the [Azure Portal](https://docs.microsoft.com/azure/postgresql/quickstart-create-server-database-portal?WT.mc_id=devto-blog-abhishgu), [Azure CLI](https://docs.microsoft.com/azure/postgresql/quickstart-create-server-database-azure-cli?WT.mc_id=devto-blog-abhishgu), [Azure PowerShell](https://docs.microsoft.com/azure/postgresql/quickstart-create-postgresql-server-database-using-azure-powershell?WT.mc_id=devto-blog-abhishgu), [ARM template](https://docs.microsoft.com/azure/postgresql/quickstart-create-postgresql-server-database-using-arm-template?tabs=azure-portal&WT.mc_id=devto-blog-abhishgu). Once you've done that, you can easily connect to the database using you favourite programming language such as [Java](https://docs.microsoft.com/azure/postgresql/connect-java?WT.mc_id=devto-blog-abhishgu), [.NET](https://docs.microsoft.com/azure/postgresql/connect-csharp?WT.mc_id=devto-blog-abhishgu), [Node.js](https://docs.microsoft.com/azure/postgresql/connect-nodejs?WT.mc_id=devto-blog-abhishgu), [Python](https://docs.microsoft.com/azure/postgresql/connect-python?WT.mc_id=devto-blog-abhishgu), [Go](https://docs.microsoft.com/azure/postgresql/connect-go?WT.mc_id=devto-blog-abhishgu) etc.

> Although the above references are for Single Server deployment mode, please note that [Hyperscale (Citus) is another deployment mode you can use](https://docs.microsoft.com/azure/postgresql/overview?WT.mc_id=devto-blog-abhishgu#azure-database-for-postgresql---hyperscale-citus) for "workloads that are approaching -- or already exceed -- 100 GB of data."

Please ensure that you keep the following PostgreSQL related information handy since you will need them to configure the Debezium Connector in the subsequent sections - database hostname (and port), username, password

### Azure Event Hubs

Azure Event Hubs is a fully managed data streaming platform and event ingestion service. [It also provides a Kafka endpoint](https://docs.microsoft.com/azure/event-hubs/event-hubs-for-kafka-ecosystem-overview?WT.mc_id=devto-blog-abhishgu) that supports Apache Kafka protocol 1.0 and later and works with existing Kafka client applications and other tools in the Kafka ecosystem including `Kafka Connect` (demonstrated in this blog).

You can use the [Azure Portal](https://docs.microsoft.com/azure/event-hubs/event-hubs-create?WT.mc_id=devto-blog-abhishgu), [Azure CLI](https://docs.microsoft.com/azure/event-hubs/event-hubs-quickstart-cli?WT.mc_id=devto-blog-abhishgu), [PowerShell](https://docs.microsoft.com/azure/event-hubs/event-hubs-quickstart-powershell?WT.mc_id=devto-blog-abhishgu) or [ARM template](https://docs.microsoft.com/azure/event-hubs/event-hubs-resource-manager-namespace-event-hub?WT.mc_id=devto-blog-abhishgu) to create an Azure Event Hubs namespace and other resources. To ensure that the Kafka functionality is enabled, all you need to do is choose the `Standard` or [`Dedicated`](https://docs.microsoft.com/azure/event-hubs/event-hubs-dedicated-overview?WT.mc_id=devto-blog-abhishgu) tier (since the Basic tier doesn't support Kafka on Event Hubs.)

After the setup, please ensure that you keep the Connection String handy since you will need it to configure Kafka Connect. You can do so [using the Azure Portal](https://docs.microsoft.com/azure/event-hubs/event-hubs-get-connection-string?WT.mc_id=devto-blog-abhishgu) or [Azure CLI](https://docs.microsoft.com/cli/azure/eventhubs/eventhub/authorization-rule/keys?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-eventhubs-eventhub-authorization-rule-keys-list)

### Install Kafka

To run Kafka Connect, I will be using a local Kafka installation just for convenience. [Just download Apache Kafka](https://kafka.apache.org/downloads), unzip its contents and you're good to go!

## Download Debezium connector and start Kafka Connect

To start with, clone this Git repo:

```bash
git clone https://github.com/abhirockzz/debezium-azure-postgres-cdc

cd debezium-azure-postgres-cdc
```

Download Debezium PostgreSQL source connector JARs

> `1.2.0` is the latest version at the time of writing

```bash
DEBEZIUM_CONNECTOR_VERSION=1.2.0

curl https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/${DEBEZIUM_CONNECTOR_VERSION}.Final/debezium-connector-postgres-${DEBEZIUM_CONNECTOR_VERSION}.Final-plugin.tar.gz --output debezium-connector-postgres.tar.gz

tar -xvzf debezium-connector-postgres.tar.gz
```

You should now see a new folder named `debezium-connector-postgres`. Copy the connector JAR files to your Kafka installation:

```
export KAFKA_HOME=[path to kafka installation e.g. /Users/foo/work/kafka_2.12-2.3.0]

cp debezium-connector-postgres/*.jar $KAFKA_HOME/libs

//confirm
ls -lrt $KAFKA_HOME/libs | grep debezium
ls -lrt $KAFKA_HOME/libs | grep protobuf
ls -lrt $KAFKA_HOME/libs | grep postgresql
```

Before starting the Kafka Connect cluster, edit the `connect.properties` file to include appropriate values for the following attributes: `bootstrap.servers`, `sasl.jaas.config`, `producer.sasl.jaas.config`, `consumer.sasl.jaas.config` (just replace the placeholders)

Start Kafka Connect cluster (I am running it in `distributed` mode):

```
export KAFKA_HOME=[path to kafka installation e.g. /Users/foo/work/kafka_2.12-2.3.0]

$KAFKA_HOME/bin/connect-distributed.sh connect.properties
```

Wait for the Kafka Connect instance to start - you should see Kafka Connect internal topics in Azure Event Hubs e.g.

![](https://dev-to-uploads.s3.amazonaws.com/i/xl53zykq2ss4iynjdufm.png)
## Configure PostgreSQL

Before installing the connector, we need to: 

- Ensure that the PostgreSQL instance is accessible from your Kafka Connect cluster 
- Ensure that the PostrgeSQL replication setting is set to "Logical"
- Create a table which you can use to try out the change data capture feature

If you're using Azure DB for PostgreSQL, create a firewall rule using [az postgres server firewall-rule create](https://docs.microsoft.com/en-us/azure/postgresql/howto-manage-firewall-using-cli#create-firewall-rule) command to whitelist your Kafka Connect host. In my case, it was a local Kafka Connect cluster, so I simply navigated to the Azure portal (**Connection security** section of my PostrgreSQL instance) and chose **Add current client IP address** to make sure that my local IP was added to the firewall rule as such:

![](pg-firewall.PNG)

To change the replication mode for Azure DB for PostgreSQL, you can use the [az postgres server configuration](https://docs.microsoft.com/cli/azure/postgres/server/configuration?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-postgres-server-configuration-set) command:

```azurecli
az postgres server configuration set --resource-group <name of resource group> --server-name <name of server> --name azure.replication_support --value logical
```

.. or use the **Replication** menu of your PostgreSQL instance in the Azure Portal:

![](pg-replication.PNG)

After updating the configuration, you will need to re-start the server which you can do using the CLI ([az postgres server restart](https://docs.microsoft.com/cli/azure/postgres/server?view=azure-cli-latest&WT.mc_id=devto-blog-abhishgu#az-postgres-server-restart)) or the portal.

Once the database is up and running, create the table - I have used `psql` CLI in this example, but feel free to use any other tool. For example, to connect to your PostgreSQL database on Azure over SSL (you will be prompted for the password):

```bash
psql -h <POSTGRESQL_INSTANCE_NAME>.postgres.database.azure.com -p 5432 -U <POSTGRES_USER_NAME> -W -d <POSTGRES_DB_NAME> --set=sslmode=require

//example
psql -h my-pgsql.postgres.database.azure.com -p 5432 -U foo@my-pgsql -W -d postgres --set=sslmode=require

//to create the table
CREATE TABLE todos (id SERIAL, description VARCHAR(30), todo_status VARCHAR(10), PRIMARY KEY(id));
```

## Install Debezium PostgreSQL source connector

Update the `pg-source-connector.json` file with the details for the Azure PostgreSQL instance. Here is an example:

```json
{
    "name": "todo-test-connector",
    "config": {
        "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
        "database.hostname": "<POSTGRES_INSTANCE_NAME>.postgres.database.azure.com",
        "database.port": "5432",
        "database.user": "<DB_USER_NAME>",
        "database.password": "<PASSWORD>",
        "database.dbname": "<DB_NAME e.g. postgres>",
        "database.server.name": "<LOGICAL_NAMESPACE e.g. todo-server>",
        "plugin.name": "wal2json",
        "table.whitelist": "<TABLE_NAMES e.g. public.todos>"
    }
}
```

Let's go through the configuration:

> For detailed info, check [Debezium documentation](https://debezium.io/documentation/reference/1.2/connectors/postgresql.html#postgresql-connector-properties)

- `connector.class`: name of the connector class (this is a static value)
- `database.hostname` and `database.port`: IP address or hostname for your PostgreSQL instance as well as the port (e.g. `5432`)
- `database.user` and `database.password`: username and password for your PostgreSQL instance
- `database.dbname`: database name e.g. `postgres`
- `database.server.name`: Logical name that identifies and provides a namespace for the particular PostgreSQL database server/cluster being monitored.
- `table.whitelist`: comma-separated list of regex specifying which tables you want to monitor for change data capture
- `plugin.name`: name of the logical decoding plug-in e.g. `wal2json`

> At the time of writing, Debezium supports the following plugins: `decoderbufs`, `wal2json`, `wal2json_rds`, `wal2json_streaming`, `wal2json_rds_streaming` and `pgoutput`. I have used `wal2json` in this example, and it's [supported on Azure as well](https://docs.microsoft.com/azure/postgresql/concepts-logical?WT.mc_id=devto-blog-abhishgu)!


Finally, install the connector!

```
curl -X POST -H "Content-Type: application/json" --data @pg-source-connector.json http://localhost:8083/connectors
```

Kafka Connect will now start monitoring the `todos` table for create, update and delete events

## Change data capture in action ...

Insert records:

```bash
psql -h <POSTGRES_INSTANCE_NAME>.postgres.database.azure.com -p 5432 -U <POSTGRES_USER_NAME> -W -d <POSTGRES_DB_NAME> --set=sslmode=require

INSERT INTO todos (description, todo_status) VALUES ('install postgresql', 'complete');
INSERT INTO todos (description, todo_status) VALUES ('install kafka', 'complete');
INSERT INTO todos (description, todo_status) VALUES ('setup source connector', 'pending');
```

The connector should now spring into action and send the CDC events to a Event Hubs topic named `<server name in config>.<table name>` e.g. `todo-server.public.todos`

Let's introspect the contents of the topic to make sure everything is working as expected. I am using [`kafkacat`](https://github.com/edenhill/kafkacat) in this example, but you can also [create a consumer app using any of these options listed here](https://docs.microsoft.com/azure/event-hubs/apache-kafka-developer-guide?WT.mc_id=devto-blog-abhishgu#quickstarts-in-github)

Update `metadata.broker.list` and `sasl.password` attributes in `kafkacat.conf` to include Kafka broker details. In a different terminal, use it to read the CDC payloads:

```bash
export KAFKACAT_CONFIG=kafkacat.conf
export BROKER=<kafka broker> e.g. for event hubs - my-eventhubs-namespace.servicebus.windows.net:9093
export TOPIC=<server config>.<table name> e.g. todo-server.public.todos

kafkacat -b $BROKER -t $TOPIC -o beginning
```

You should see the JSON payloads representing the change data events generated in PostgreSQL in response to the rows you had just added to the `todos` table. Here is a snippet of the payload:


```json
{
    "schema": {...},
    "payload": {
        "before": null,
        "after": {
            "id": 1,
            "description": "install postgresql",
            "todo_status": "complete"
        },
        "source": {
            "version": "1.2.0.Final",
            "connector": "postgresql",
            "name": "fullfillment",
            "ts_ms": 1593018069944,
            "snapshot": "last",
            "db": "postgres",
            "schema": "public",
            "table": "todos",
            "txId": 602,
            "lsn": 184579736,
            "xmin": null
        },
        "op": "c",
        "ts_ms": 1593018069947,
        "transaction": null
    }
```

The event consists of the `payload` along with its `schema` (omitted for brevity). In `payload` section, notice how the create operation (`"op": "c"`) is represented - `"before": null` means that this was a newly `INSERT`ed row, `after` provides values for the each columns in the row, `source` provides the PostgreSQL instance metadata from where this event was picked up etc.

You can try the same with update or delete operations as well and introspect the CDC events, e.g.

```sql
UPDATE todos SET todo_status = 'complete' WHERE description = 'setup source connector';
```

## (Optional) Install File sink connector

As bonus, you can quickly test this with a File Sink connector as well. It is available out of the box in the Kafka distribution - all you need to do is install the connector. Just replace the `topics` and `file` attribute in `file-sink-connector.json` file

```json
{
    "name": "cdc-file-sink",
    "config": {
        "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
        "tasks.max": "1",
        "topics": "<server name>.<table name> e.g. todos-server.public.todos",
        "file": "<enter full path to file e.g. /Users/foo/work/pg-cdc.txt>"
    }
}
```

To create the connector:

```bash
curl -X POST -H "Content-Type: application/json" --data @file-sink-connector.json http://localhost:8083/connectors
```

Play around with the database records and monitor the records in the configured output sink file, e.g.

```bash
tail -f /Users/foo/work/pg-cdc.txt
```

## Conclusion

If you've reached this far, thanks for reading (this rather lengthy tutorial)!

![](https://media.giphy.com/media/3osxYdXvsGw6wT5lIY/giphy.gif)

Change Data Capture is a powerful technique which can help "unlock the database" by providing near real-time access to it's changes. This was a "getting started" guide meant to help you get up and running quickly, experiment with and iterate further. Hope you found it useful!