CDC from Postgres with AWS MSK and Debezium

Objective:

We want to create a CDC data pipeline from RDS Postgres using the Debezium Postgres source connector which will capture all the data events from the mentioned tables and pass on them to the designated topics for downstream application processing.

We need to execute below steps:

Part 1: Launch MSK cluster

Part 2: Launch a Kafka admin client.

Part 3: DB changes

Part 4: Provision a Connector

Part5: Validation

Part 1: Launching MSK cluster.

We will be using a bare minimum configuration to avoid the costing concerns. Will create a new VPC and private subnets to host MSK and later on do the VPC peering between MSK VPC and VPC where our applications are hosted.

Step 1.1:

Go to AWS console -> Search MSK -> Create cluster then In Cluster Settings

• Select cluster creation method as ‘custom create’
• Cluster type as ‘Provisioned’
• Broker type -> t3.small , No. of Zones -> 3 and Storage as 100Gi
• Select MSK default configuration.

Step 1.2: In ‘Networking Section’ select your existing VPC.

• Select zones and subnets.
• Create a new Security group with new name i.e. MSK-SG-01 and select your VPC with all default IN/OUT bound rules (we can edit them later)
• Now browse for your SG and delete the existing SG.

Step 1.3: In ‘Security Section’

• Select ACL as ‘Unauthenticated access’ and ‘IAM’ (optional)
• Encryption as ‘Plaintext’ and ‘TLS’ both
• Rest all default settings.

Step 1.4: In ‘Monitoring Section’

• All default and create the cluster.

Part 2: Launch a Kafka admin client.

Since we will be creating that in the private subnet hence it won’t be accessible with public IP. If you have a bastion box set up for your infrastructure than you can SSH into from there to perform below activities

• Create a EC2 instance with in the VPC where your MSK is hosted.
• Create a new SG with all default settings.
• Create Instance
• Edit the MSK SG inbound rule to receive the traffic from the EC2 instance. GOTO MSK dashboard -> select properties -> edit Inbound -> Type All traffic -> Source your ec2 instance -> Save.

Step 2.1: Setting the AWS CLI to interact with Kafka infra:

// configure aws cli
aws configure

// now give your region i.e 'ap-south-1'
// List clusters
aws kafka list-clusters
// Get bootstrap broker details:
aws kafka get-bootstrap-brokers --cluster-arn <<cluster-arn>>
// Describe cluster:
aws kafka describe-cluster --cluster-arn <<cluster-arn>>

Step 2.2: Setting the Kafka on EC2 to validate and do the operations.

// set up java
sudo yum install java-1.8.0
// download kafka 
wget https://archive.apache.org/dist/kafka/3.4.1/kafka_2.12-3.4.1.tgz
// untar the kafka zip
tar -xzf kafka_2.12-3.4.1.tgz -C ./kafka

Step 2.3: Custom Kafka config for auto creation topics.

Create a file with name “kafka-topic-auto-create-config.txt” with below config.

we have to be bit careful while giving the file path as it warrants two forward slash and then file path i.e fileb:///home/ec2-user/kafka-topic-auto-create-config.txt

Kafka related commands:

Part3: DB changes.

Assuming our RDS Postgres instance in not public and can be accessed only from the private subnets.

1. 1. Go to Parameter Groups from the RDS Dashboard and click Create Parameter Group. [we can choose to copy and create the same parameter group already applied]
   2. Name this parameter group something like cdc-enabled-postgres-group, locate the option with the key rds.logical_replication and set the value to 1. Click Create Parameter Group.
   3. If your DB instance status has transitioned to ‘available’, click Modify.
   4. Scroll to Additional Configuration, set the initial database name to Postgres and switch the parameter group to the one you just created.
   5. Schedule the modifications to take effect immediately and click Modify Instance. Your instance state will transition to ‘Modifying’.
   6. Once the instance is modified, you’ll need to reboot it for the rds.logical_replication setting to update. Select your instance, then click Actions > Reboot.
   7. Edit the RDS inbound rules to allow the request from MSK VPC

To verify that your wal_level property is set to logical by running the following command from any PG client:

Creating a ‘Debezium’ role for our connector with proper privilege’s:

Step 4.3: Create a S3 bucket for custom plugin.

Step 4.4: Create a custom worker configuration with information about your configuration provider.

custom-kafkaconnect-policy

Step 4.6: Finally create a Debezium Postgres connector.

1. Choose the plugin.
2. Choose the MSK cluster with NONE authentication type.
3. Provide below configuration [Debezium postgres connector config]
4. Connector capacity → provisioned
5. Worker configuration → default <or provide the custom one created as above>
6. AccessPermission → Custom-AwsRoleforKafkaConnect
7. Create a cloud watch log group for the logs [It’ll help to debug the issues]

Debezium postgres connector config

Part 5: Validation

• To validate the flow, you have to log in into the EC2 Kafka admin client.
• List the topic which shall be already created already with the naming convention as postgres.<schema-name>.<table-name>
• check with the message with below command.

Conclusion:

Congratulations! your CDC pipeline is now created.