Debezium Server with RabbitMQ

Overview

Debezium Server: Debezium Server is a lightweight runtime that can stream changes from various databases to a range of downstream systems like message brokers, databases, or other services. It is an alternative to Kafka Connect and allows the use of different messaging systems.

Configuration

Debezium Server uses Micro-Profile Configuration for configuration. This means that the application can be configured from disparate sources like configuration files, environment variables, system properties etc.

The main configuration file is conf/application.properties.

# Sink connector config - RabbitMQ
debezium.sink.type=rabbitmq
debezium.sink.rabbitmq.connection.host=localhost
debezium.sink.rabbitmq.connection.port=5672
debezium.sink.rabbitmq.connection.username=guest
debezium.sink.rabbitmq.connection.password=guest
debezium.sink.rabbitmq.connection.virtual.host=vhost
debezium.sink.rabbitmq.ackTimeout=3000
debezium.sink.rabbitmq.exchange=test_topic_exchange
debezium.sink.rabbitmq.routingKey=test_topic_routing_key

# Source connector config - PostgreSQL
debezium.source.connector.class=io.debezium.connector.postgresql.PostgresConnector
debezium.source.plugin.name=pgoutput
debezium.source.offset.storage.file.filename=data/offsets.dat
debezium.source.offset.flush.interval.ms=0
debezium.source.database.hostname=localhost
debezium.source.database.port=5433
debezium.source.database.user=postgres
debezium.source.database.password=root
debezium.source.database.dbname=source_db
debezium.source.topic.prefix=test_topic_prefix

# Format config
debezium.format.key=json
debezium.format.value=json

# Quarkus
quarkus.log.console.json=false

Docker compose:

services:

  debezium:
    image: quay.io/debezium/server:2.5
    container_name: debezium
    healthcheck:
      test: curl http://debezium:8997/q/health || exit 1
      interval: 5s
      timeout: 5s
      retries: 5
    ports:
      - "8997:8997"
    volumes:
      - ./debezium_conf:/debezium/conf

Running RabbitMQ as a Docker container

docker run -it --rm --name rabbitmq -p 5672:5672 -p 15672:15672 rabbitmq:3.13-management

Consumer Service:

A separate application/service that reads changes from RabbitMQ and writes them to the destination database.

To achieve Change Data Capture (CDC) when the destination is another database using Debezium Server and RabbitMQ, you need to set up the process where the changes captured from the source database (PostgreSQL) are published to RabbitMQ, and then a consumer service reads these changes from RabbitMQ and writes them to the destination database.

Here's a sample Java code for a consumer service that reads change events from RabbitMQ and writes them to a destination PostgreSQL database.

import com.rabbitmq.client.*;
import org.json.JSONObject;

import java.io.IOException;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.util.concurrent.TimeoutException;

public class RabbitMQConsumer {
    private static final String RABBITMQ_HOST = "localhost";
    private static final int RABBITMQ_PORT = 5672;
    private static final String RABBITMQ_USERNAME = "guest";
    private static final String RABBITMQ_PASSWORD = "guest";
    private static final String RABBITMQ_VIRTUAL_HOST = "vhost";
    private static final String EXCHANGE_NAME = "test_topic_exchange";
    private static final String ROUTING_KEY = "test_topic_routing_key";
    private static final String QUEUE_NAME = "cdc_queue";

    private static final String DB_URL = "jdbc:postgresql://localhost:5432/destination_db";
    private static final String DB_USER = "postgres";
    private static final String DB_PASSWORD = "root";

    public static void main(String[] args) throws IOException, TimeoutException {
        ConnectionFactory factory = new ConnectionFactory();
        factory.setHost(RABBITMQ_HOST);
        factory.setPort(RABBITMQ_PORT);
        factory.setUsername(RABBITMQ_USERNAME);
        factory.setPassword(RABBITMQ_PASSWORD);
        factory.setVirtualHost(RABBITMQ_VIRTUAL_HOST);

        Connection connection = factory.newConnection();
        Channel channel = connection.createChannel();

        channel.exchangeDeclare(EXCHANGE_NAME, "topic", true);
        channel.queueDeclare(QUEUE_NAME, true, false, false, null);
        channel.queueBind(QUEUE_NAME, EXCHANGE_NAME, ROUTING_KEY);

        DeliverCallback deliverCallback = (consumerTag, delivery) -> {
            String message = new String(delivery.getBody(), "UTF-8");
            System.out.println("Received: " + message);

            JSONObject jsonObject = new JSONObject(message);
            long id = jsonObject.getLong("id");
            String firstName = jsonObject.getString("first_name");
            String lastName = jsonObject.getString("last_name");

            try (Connection dbConnection = DriverManager.getConnection(DB_URL, DB_USER, DB_PASSWORD)) {
                String sql = "INSERT INTO destination_table (id, first_name, last_name) VALUES (?, ?, ?) ON CONFLICT (id) DO UPDATE SET first_name = EXCLUDED.first_name, last_name = EXCLUDED.last_name";
                try (PreparedStatement statement = dbConnection.prepareStatement(sql)) {
                    statement.setLong(1, id);
                    statement.setString(2, firstName);
                    statement.setString(3, lastName);
                    statement.executeUpdate();
                }
            } catch (SQLException e) {
                e.printStackTrace();
            }
        };

        channel.basicConsume(QUEUE_NAME, true, deliverCallback, consumerTag -> {});
    }
}

Conclusion

This guide provides an overview of using Debezium Server with RabbitMQ for CDC, highlighting key concepts, configuration properties, and use cases to help you get started with this powerful combination for real-time data streaming and processing.

References

• Debezium Server Documentation

• Debezium Documentation