Change Data Capture Pipelines with Debezium 3.x and Kafka Connect

Introduction

In modern distributed systems, keeping data in sync across multiple services is one of the hardest problems architects face. Traditional approaches — polling databases, writing custom triggers, building ETL jobs — all come with significant drawbacks: high latency, tight coupling, performance overhead, or blind spots in the change history.

Change Data Capture (CDC) offers a fundamentally different approach. Instead of asking "what's in the database now?", CDC asks "what changed, and when?" By reading directly from the database's transaction log (WAL in PostgreSQL, binlog in MySQL), CDC captures every insert, update, and delete as it happens — with zero overhead on the source database — and publishes those changes as a stream of events.

Debezium, the leading open-source CDC platform, has matured significantly with its 3.x series. Paired with Kafka Connect as the distributed runtime, it forms the backbone of countless real-time data pipelines at scale. In this tutorial, you'll build a production-ready CDC pipeline that captures changes from a PostgreSQL orders table and makes them available to downstream Spring Boot consumers.

What you'll build:

• A PostgreSQL database with logical replication enabled


• Kafka and Kafka Connect running via Docker Compose


• A Debezium 3.x PostgreSQL connector capturing order changes


• A Spring Boot 3.4 consumer processing CDC events


• Integration tests using Testcontainers

Prerequisites

• Java 21+
• Docker and Docker Compose
• Maven 3.9+
• Basic familiarity with Kafka and Spring Boot

Tool versions used in this tutorial:

• Debezium 3.6.2.Final


• Apache Kafka 3.9.0


• Kafka Connect 3.9.0 (bundled with Kafka)


• Spring Boot 3.4.5


• PostgreSQL 16

Architecture Overview

The CDC pipeline follows this data flow:

PostgreSQL (WAL)
     │
     ▼ (logical replication slot)
Debezium PostgreSQL Connector
     │
     ▼ (Kafka Connect)
Kafka Topic: orders.public.orders
     │
     ▼ (Kafka Consumer)
Spring Boot Service
     │
     ▼ (business logic)
Read Model / Search Index / Audit Log

Debezium acts as a PostgreSQL replication client. It creates a replication slot in PostgreSQL, which holds the WAL position Debezium has processed. This ensures no events are lost even if Debezium restarts — Kafka Connect stores the offset in Kafka itself, and Debezium re-reads from the replication slot on recovery.

Step 1: PostgreSQL Configuration

Logical replication requires specific PostgreSQL settings. Create docker/postgres/postgresql.conf:

wal_level = logical
max_replication_slots = 5
max_wal_senders = 5

Create the initialization SQL in docker/postgres/init.sql:

-- Application user
CREATE USER app_user WITH PASSWORD 'app_password';
CREATE DATABASE orders_db OWNER app_user;

-- Dedicated replication user with minimal privileges
CREATE USER debezium WITH REPLICATION PASSWORD 'debezium_password';

\c orders_db;

GRANT USAGE ON SCHEMA public TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;

-- Application table
CREATE TABLE IF NOT EXISTS orders (
    id BIGSERIAL PRIMARY KEY,
    customer_id BIGINT NOT NULL,
    product_code VARCHAR(50) NOT NULL,
    quantity INT NOT NULL,
    total_amount DECIMAL(10, 2) NOT NULL,
    status VARCHAR(20) NOT NULL DEFAULT 'PENDING',
    created_at TIMESTAMP WITH TIME ZONE DEFAULT NOW(),
    updated_at TIMESTAMP WITH TIME ZONE DEFAULT NOW()
);

-- Named publication is safer than FOR ALL TABLES:
-- limits replication scope and makes schema changes deliberate.
CREATE PUBLICATION dbz_publication FOR TABLE orders;

Step 2: Docker Compose Environment

version: '3.9'

services:
  postgres:
    image: postgres:16-alpine
    environment:
      POSTGRES_PASSWORD: admin_password
    ports:
      - "5432:5432"
    volumes:
      - ./docker/postgres/postgresql.conf:/etc/postgresql/postgresql.conf
      - ./docker/postgres/init.sql:/docker-entrypoint-initdb.d/init.sql
    command: postgres -c config_file=/etc/postgresql/postgresql.conf

  zookeeper:
    image: confluentinc/cp-zookeeper:7.7.0
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000

  kafka:
    image: confluentinc/cp-kafka:7.7.0
    depends_on: [zookeeper]
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_AUTO_CREATE_TOPICS_ENABLE: 'false'

  kafka-connect:
    image: debezium/connect:3.0
    depends_on: [kafka, postgres]
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:29092
      GROUP_ID: 1
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-status
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1

The official debezium/connect:3.0 image packages Kafka Connect with Debezium connectors pre-installed. In production, consider Strimzi or a custom image containing only the connectors you need.

Step 3: Registering the Debezium Connector

With the stack running, register the connector via Kafka Connect's REST API. Create connector-config/orders-connector.json:

{
  "name": "orders-postgres-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "debezium_password",
    "database.dbname": "orders_db",
    "database.server.name": "orders-server",
    "topic.prefix": "orders",
    "plugin.name": "pgoutput",
    "publication.name": "dbz_publication",
    "publication.autocreate.mode": "disabled",
    "table.include.list": "public.orders",
    "snapshot.mode": "initial",
    "decimal.handling.mode": "double",
    "time.precision.mode": "connect",
    "tombstones.on.delete": "true",
    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.unwrap.add.fields": "op,source.ts_ms,before",
    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "false",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "false"
  }
}

Register it:

curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @connector-config/orders-connector.json

# Verify status
curl http://localhost:8083/connectors/orders-postgres-connector/status

Key configuration choices:

• plugin.name: pgoutput — the native PostgreSQL logical decoding plugin (available since PG 10), requiring no separate installation. wal2json was deprecated in Debezium 1.x and removed in 2.0; never use it for new pipelines.
• publication.autocreate.mode: disabled — Debezium uses the publication you created in init.sql instead of attempting to create one (which requires superuser privileges).
• snapshot.mode: initial — takes a consistent snapshot of existing rows on first run, then switches to streaming. Use never if you only care about changes going forward.
• transforms: unwrap with ExtractNewRecordState — strips Debezium's envelope format and presents just the row fields plus metadata. This dramatically simplifies consumers.
• decimal.handling.mode: double — avoids the Debezium-specific DECIMAL struct for monetary fields. Use string if you need exact precision.

Step 4: Spring Boot Consumer

Project Structure

src/main/java/com/vimleshpandey/demo/
├── CdcPipelineApplication.java
├── cdc/
│   ├── OrderEventConsumer.java
│   ├── config/
│   │   └── KafkaConfig.java
│   ├── model/
│   │   └── OrderChangeEvent.java
│   └── service/
│       └── OrderSyncService.java

The Change Event Model

With ExtractNewRecordState, the Kafka message value looks like:

{
  "id": 1,
  "customer_id": 42,
  "product_code": "PROD-001",
  "quantity": 2,
  "total_amount": 149.98,
  "status": "PENDING",
  "__op": "c",
  "__source_ts_ms": 1730000000000,
  "__before": null
}

Map this to a Java record:

package com.vimleshpandey.demo.cdc.model;

import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
import com.fasterxml.jackson.annotation.JsonProperty;
import java.math.BigDecimal;
import java.time.Instant;

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
    Long id,
    @JsonProperty("customer_id") Long customerId,
    @JsonProperty("product_code") String productCode,
    Integer quantity,
    @JsonProperty("total_amount") BigDecimal totalAmount,
    String status,
    @JsonProperty("created_at") Instant createdAt,
    @JsonProperty("updated_at") Instant updatedAt,
    String __op,
    @JsonProperty("__source_ts_ms") Long sourceTimestampMs,
    @JsonProperty("__before") OrderSnapshot before
) {

    public enum Operation {
        CREATE, UPDATE, DELETE, READ;

        public static Operation fromDebezium(String code) {
            return switch (code) {
                case "c" -> CREATE;
                case "u" -> UPDATE;
                case "d" -> DELETE;
                case "r" -> READ;
                default -> throw new IllegalArgumentException("Unknown op: " + code);
            };
        }
    }

    public Operation operation() {
        return Operation.fromDebezium(__op);
    }

    @JsonIgnoreProperties(ignoreUnknown = true)
    public record OrderSnapshot(
        Long id,
        String status,
        @JsonProperty("total_amount") BigDecimal totalAmount
    ) {}
}

Kafka Configuration

package com.vimleshpandey.demo.cdc.config;

import com.vimleshpandey.demo.cdc.model.OrderChangeEvent;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.listener.ContainerProperties;
import org.springframework.kafka.support.serializer.JsonDeserializer;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class KafkaConfig {

    @Value("${spring.kafka.bootstrap-servers}")
    private String bootstrapServers;

    @Bean
    public ConsumerFactory<String, OrderChangeEvent> orderEventConsumerFactory() {
        JsonDeserializer<OrderChangeEvent> deserializer =
            new JsonDeserializer<>(OrderChangeEvent.class);
        deserializer.setUseTypeHeaders(false);

        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "order-cdc-consumer");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);

        return new DefaultKafkaConsumerFactory<>(
            props, new StringDeserializer(), deserializer);
    }

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, OrderChangeEvent>
    orderEventListenerContainerFactory(
        ConsumerFactory<String, OrderChangeEvent> factory) {

        var cf = new ConcurrentKafkaListenerContainerFactory<String, OrderChangeEvent>();
        cf.setConsumerFactory(factory);
        cf.getContainerProperties().setAckMode(ContainerProperties.AckMode.MANUAL_IMMEDIATE);
        return cf;
    }
}

The Event Consumer

package com.vimleshpandey.demo.cdc;

import com.vimleshpandey.demo.cdc.model.OrderChangeEvent;
import com.vimleshpandey.demo.cdc.service.OrderSyncService;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

@Component
public class OrderEventConsumer {

    private static final Logger log = LoggerFactory.getLogger(OrderEventConsumer.class);
    private final OrderSyncService syncService;

    public OrderEventConsumer(OrderSyncService syncService) {
        this.syncService = syncService;
    }

    @KafkaListener(
        topics = "orders.public.orders",
        containerFactory = "orderEventListenerContainerFactory"
    )
    public void onOrderChange(
        ConsumerRecord<String, OrderChangeEvent> record, Acknowledgment ack) {

        OrderChangeEvent event = record.value();

        // Tombstone records (delete markers on compacted topics) have null value
        if (event == null) {
            log.debug("Tombstone for key={}", record.key());
            ack.acknowledge();
            return;
        }

        try {
            switch (event.operation()) {
                case CREATE, READ -> syncService.upsertOrder(event);
                case UPDATE       -> syncService.handleOrderUpdate(event);
                case DELETE       -> syncService.deleteOrder(event.id());
            }
            ack.acknowledge();
        } catch (Exception e) {
            log.error("Failed to process CDC event order id={}", event.id(), e);
            throw e; // Do NOT ack — let the container retry via its error handler
        }
    }
}

The Sync Service

package com.vimleshpandey.demo.cdc.service;

import com.vimleshpandey.demo.cdc.model.OrderChangeEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

@Service
public class OrderSyncService {

    private static final Logger log = LoggerFactory.getLogger(OrderSyncService.class);

    public void upsertOrder(OrderChangeEvent event) {
        log.info("Upserting order id={} status={} amount={}",
            event.id(), event.status(), event.totalAmount());
        // Write to read model, search index, analytics store, etc.
    }

    public void handleOrderUpdate(OrderChangeEvent event) {
        if (event.before() != null
                && !event.before().status().equals(event.status())) {
            log.info("Order {} status changed: {} -> {}",
                event.id(), event.before().status(), event.status());
        }
        upsertOrder(event);
    }

    public void deleteOrder(Long orderId) {
        log.info("Removing order id={} from read model", orderId);
    }
}

Step 5: Understanding the Event Envelope

Without ExtractNewRecordState, Debezium's raw message looks like:

{
  "before": null,
  "after": {
    "id": 1,
    "customer_id": 42,
    "product_code": "PROD-001",
    "quantity": 2,
    "total_amount": 149.98,
    "status": "PENDING"
  },
  "source": {
    "version": "3.6.2.Final",
    "connector": "postgresql",
    "name": "orders-server",
    "ts_ms": 1730000000000,
    "lsn": 12345678
  },
  "op": "c",
  "ts_ms": 1730000000100
}

The SMT flattens after to the top level and adds __op, __source_ts_ms, and __before fields. This simplifies deserialization but also discards the raw source block — if you need fields like lsn (log sequence number) for idempotency tracking, add them to transforms.unwrap.add.fields.

For UPDATE events, the __before field contains the row's previous state (only the columns Debezium tracks — by default all). This is invaluable for detecting which fields actually changed without maintaining your own shadow state.

Step 6: Testing with Testcontainers

package com.vimleshpandey.demo;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.junit.jupiter.api.Test;
import org.testcontainers.containers.KafkaContainer;
import org.testcontainers.containers.PostgreSQLContainer;
import org.testcontainers.junit.jupiter.Container;
import org.testcontainers.junit.jupiter.Testcontainers;
import org.testcontainers.utility.DockerImageName;

import java.time.Duration;
import java.util.List;
import java.util.Properties;

import static org.assertj.core.api.Assertions.assertThat;

@Testcontainers
class CdcPipelineIntegrationTest {

    @Container
    static PostgreSQLContainer<?> postgres =
        new PostgreSQLContainer<>("postgres:16-alpine")
            .withInitScript("init.sql")
            .withCommand(
                "postgres",
                "-c", "wal_level=logical",
                "-c", "max_replication_slots=5",
                "-c", "max_wal_senders=5"
            );

    @Container
    static KafkaContainer kafka =
        new KafkaContainer(DockerImageName.parse("confluentinc/cp-kafka:7.7.0"));

    @Test
    void consumerCanPollKafkaTopic() {
        // In a full integration test you would:
        // 1. Register the Debezium connector via REST, pointing at the Testcontainer Postgres
        // 2. Insert a row into the orders table
        // 3. Poll the topic and assert event fields
        // Debezium's EmbeddedEngine (debezium-api) is the cleanest way to do this
        // without running a full Kafka Connect cluster in the test.

        try (KafkaConsumer<String, String> consumer = buildConsumer()) {
            consumer.subscribe(List.of("orders.public.orders"));
            var records = consumer.poll(Duration.ofSeconds(5));
            assertThat(records).isNotNull();
        }
    }

    private KafkaConsumer<String, String> buildConsumer() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, kafka.getBootstrapServers());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "test-group");
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
        return new KafkaConsumer<>(props);
    }
}

Running the Application

# Start infrastructure
docker-compose up -d postgres zookeeper kafka kafka-connect

# Wait for Kafka Connect REST API to become available
until curl -sf http://localhost:8083/connectors; do
  echo "Waiting for Kafka Connect..."; sleep 3
done

# Register the connector
curl -X POST http://localhost:8083/connectors \
  -H "Content-Type: application/json" \
  -d @connector-config/orders-connector.json

# Verify connector and task are RUNNING
curl http://localhost:8083/connectors/orders-postgres-connector/status

# Insert a test order
docker exec $(docker-compose ps -q postgres) \
  psql -U app_user -d orders_db \
  -c "INSERT INTO orders (customer_id, product_code, quantity, total_amount, status)
      VALUES (42, 'PROD-001', 2, 149.98, 'PENDING');"

# Watch the CDC topic
docker exec $(docker-compose ps -q kafka) \
  kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 \
  --topic orders.public.orders \
  --from-beginning

# Start the Spring Boot consumer
./mvnw spring-boot:run

Common Pitfalls and Tips

1. Replication slot accumulation

Every Debezium connector instance creates a replication slot. If a connector is deleted without dropping its slot, PostgreSQL keeps buffering WAL indefinitely, growing disk usage until the server crashes. Monitor and clean up:

SELECT slot_name, active, restart_lsn, pg_size_pretty(
    pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)
) AS lag
FROM pg_replication_slots;

-- Drop an orphaned slot
SELECT pg_drop_replication_slot('orders-server');

2. Snapshot mode choices

initial is safe but slow on large tables. For millions of rows, consider running with initial_only first, then changing to no_data on subsequent restarts. Debezium 3.x's incremental snapshots (snapshot.mode: when_needed) re-snapshot only tables that were offline during streaming, which avoids full re-snapshots after brief outages.

3. Consumer idempotency is mandatory

Debezium guarantees at-least-once delivery. Your consumer must handle duplicate events. Using the id as a natural key and doing upserts (not blind inserts) in your read model achieves this cheaply.

4. Schema changes

Adding a nullable column or a new table to the publication is safe with a connector restart. Renaming or dropping a column requires coordinating the change with a new snapshot — plan it as a maintenance window and test in staging first.

5. Monitoring lag

Track the Kafka consumer group lag on your CDC topic. A growing lag means the consumer is falling behind. Check for slow downstream writes, insufficient consumer instances, or a max.poll.records that's set too high relative to processing time.

6. Don't point at a read replica

Despite Debezium 3.4+ supporting read replicas for incremental snapshots, the logical replication slot lives on the primary. The connector must connect to the primary for streaming; routing it to a replica is not supported.

Conclusion

You've built a complete CDC pipeline with Debezium 3.x and Kafka Connect: a PostgreSQL source database with logical replication, a connector using the native pgoutput plugin and ExtractNewRecordState SMT for simplified event shapes, and an idempotent Spring Boot consumer that handles inserts, updates, and deletes.

CDC unlocks use cases that polling can't match: real-time search indexing, audit trails that can never miss an event, cache invalidation keyed to actual database changes, and cross-service data sync without tight coupling. The same pattern applies to MySQL (via the Debezium MySQL connector, using the binlog) and MongoDB — once you understand the PostgreSQL setup, other sources are mostly connector configuration.

For production hardening, focus next on: monitoring replication slot lag, alerting on connector state transitions, credential rotation without downtime, and a clear schema evolution strategy — Avro with Schema Registry if you have multiple consumer teams, JSON with strict discipline if you control all consumers.