Change Data Capture with Debezium and Kafka

1. Overview

When a service needs to know that data in another database has just changed, a common approach is to write to the database and then send an additional message to Kafka. This approach is called dual-write and can easily lead to data inconsistency. Change Data Capture (CDC) solves this problem by reading the database’s transaction log directly.

In this article, you’ll set up PostgreSQL, Kafka, and Debezium using Docker Compose, then register a connector. After that, you’ll write a Spring Boot consumer to handle INSERT, UPDATE, and DELETE events on the orders table.

2. Why You Need CDC

Take a look at this familiar dual-write code:

@Transactional
public void placeOrder(Order order) {
    orderRepository.save(order);
    kafkaTemplate.send("orders", order.getId().toString(), toJson(order));
}

This code has two weaknesses:

• If Kafka fails after save succeeds, the database has the order but no event is sent.
• If send succeeds but the transaction rolls back, the consumer receives an event for an order that doesn’t exist.

CDC reverses this flow. The application only writes to the database. A separate process reads the transaction log (WAL for PostgreSQL, binlog for MySQL) and emits an event for each committed change. Since the log only contains committed changes, events always match the real data.

Debezium is a set of CDC connectors that run on Kafka Connect. The data flow is as follows:

Application ──write──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Kafka topic ──> Consumer

3. Setting Up the Environment

3.1. Docker Compose

You need three containers: PostgreSQL with logical replication enabled, Kafka running in KRaft mode, and Kafka Connect with the Debezium plugin already installed.

services:
  postgres:
    image: postgres:16
    command: ["postgres", "-c", "wal_level=logical"]
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: shop
    ports:
      - "5432:5432"

  kafka:
    image: apache/kafka:3.8.0
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: PLAINTEXT://:29092,CONTROLLER://:9093,EXTERNAL://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,EXTERNAL://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  connect:
    image: quay.io/debezium/connect:2.7
    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_statuses

There are three important points to note in this file:

• wal_level=logical is mandatory. Without this parameter, Debezium cannot create a replication slot.
• Kafka has two listeners. Containers use kafka:29092, while your Spring Boot application running on your machine uses localhost:9092.
• Kafka Connect stores configuration, offsets, and status in three separate topics. This way, when the connector restarts, it can resume reading from its previous position in the WAL.

Start the services and create the table:

docker compose up -d
docker compose exec postgres psql -U postgres -d shop -c "
  CREATE TABLE orders (
    id BIGSERIAL PRIMARY KEY,
    customer_email VARCHAR(255) NOT NULL,
    status VARCHAR(32) NOT NULL,
    total NUMERIC(10,2) NOT NULL
  );
  ALTER TABLE orders REPLICA IDENTITY FULL;"

REPLICA IDENTITY FULL requires PostgreSQL to write the entire old row to the WAL. Without this command, the before field in UPDATE events will be null, and DELETE events will only contain the primary key.

3.2. Registering the Connector

Kafka Connect provides a REST API on port 8083. You register the PostgreSQL connector like this:

curl -X POST -H "Content-Type: application/json" localhost:8083/connectors -d '{
  "name": "shop-connector",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "plugin.name": "pgoutput",
    "database.hostname": "postgres",
    "database.port": "5432",
    "database.user": "postgres",
    "database.password": "postgres",
    "database.dbname": "shop",
    "topic.prefix": "shop",
    "table.include.list": "public.orders",
    "slot.name": "shop_slot",
    "publication.autocreate.mode": "filtered",
    "decimal.handling.mode": "string",
    "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"
  }
}'

The meaning of important properties:

• plugin.name: pgoutput uses PostgreSQL 10+’s built-in decoding plugin, so you don’t need to install an additional extension.
• topic.prefix combined with the schema and table name creates the topic name. Here the topic is shop.public.orders.
• table.include.list limits which tables are captured.
• decimal.handling.mode: string sends NUMERIC columns as strings like "149.90". By default, Debezium encodes these values as base64 bytes, which are very hard to read.
• schemas.enable: false removes the schema section that comes with each JSON message, making the payload more compact.

Check the connector status:

curl -s localhost:8083/connectors/shop-connector/status

{"name":"shop-connector","connector":{"state":"RUNNING","worker_id":"172.18.0.4:8083"},
 "tasks":[{"id":0,"state":"RUNNING","worker_id":"172.18.0.4:8083"}],"type":"source"}

4. Change Event Structure

Add and then update an order:

docker compose exec postgres psql -U postgres -d shop -c "
  INSERT INTO orders (customer_email, status, total) VALUES ('an@example.com', 'NEW', 149.90);
  UPDATE orders SET status = 'PAID' WHERE id = 1;"

Then read the topic using the console consumer available in the Kafka image:

docker compose exec kafka /opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server localhost:9092 --topic shop.public.orders --from-beginning

An UPDATE event looks like this (with the source field abbreviated):

{
  "before": {"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
  "after":  {"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
  "source": {"connector":"postgresql","db":"shop","table":"orders","lsn":24023128, ...},
  "op": "u",
  "ts_ms": 1760170000123
}

The op field indicates the type of change:

op	Meaning	before	after
c	INSERT	null	new row
u	UPDATE	old row	new row
d	DELETE	old row	null
r	Read during initial snapshot	null	current row

When deleting a row, Debezium sends an additional tombstone, which is a message with the same key but a null value. Tombstones allow Kafka’s log compaction to completely clean up that key. Your consumer must handle null values.

5. Consuming Events with Spring Boot

5.1. Dependencies and Configuration

<dependencies>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-json</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.kafka</groupId>
        <artifactId>spring-kafka</artifactId>
    </dependency>
    <dependency>
        <groupId>org.springframework.boot</groupId>
        <artifactId>spring-boot-starter-test</artifactId>
        <scope>test</scope>
    </dependency>
</dependencies>

The versions of these dependencies are managed by spring-boot-starter-parent 3.x.

spring:
  kafka:
    bootstrap-servers: localhost:9092
    consumer:
      group-id: order-sync
      auto-offset-reset: earliest
      key-deserializer: org.apache.kafka.common.serialization.StringDeserializer
      value-deserializer: org.apache.kafka.common.serialization.StringDeserializer

Here the consumer receives the value as a String and parses it itself. This approach gives you better control over tombstones and JSON errors compared to using JsonDeserializer.

5.2. Models and Parser

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderRow(
        Long id,
        @JsonProperty("customer_email") String customerEmail,
        String status,
        BigDecimal total) {
}

@JsonIgnoreProperties(ignoreUnknown = true)
public record OrderChangeEvent(
        OrderRow before,
        OrderRow after,
        String op,
        @JsonProperty("ts_ms") Long tsMs) {
}

@JsonIgnoreProperties(ignoreUnknown = true) ignores the source field and any new fields Debezium might add later. Jackson automatically converts the string "149.90" to BigDecimal, so you don’t lose precision like you would with double.

@Component
public class OrderChangeParser {

    private final ObjectMapper objectMapper;

    public OrderChangeParser(ObjectMapper objectMapper) {
        this.objectMapper = objectMapper;
    }

    public Optional<OrderChangeEvent> parse(String value) {
        if (value == null) {
            return Optional.empty(); // tombstone
        }
        try {
            return Optional.of(objectMapper.readValue(value, OrderChangeEvent.class));
        } catch (JsonProcessingException e) {
            throw new IllegalArgumentException("Invalid Debezium event", e);
        }
    }
}

The parser returns Optional.empty() for tombstones, so the listener can simply skip them.

5.3. Listener

@Component
public class OrderCdcListener {

    private static final Logger log = LoggerFactory.getLogger(OrderCdcListener.class);
    private final OrderChangeParser parser;

    public OrderCdcListener(OrderChangeParser parser) {
        this.parser = parser;
    }

    @KafkaListener(topics = "shop.public.orders")
    public void onOrderChange(@Payload(required = false) String value) {
        parser.parse(value).ifPresent(event -> {
            switch (event.op()) {
                case "c", "r" -> log.info("Order created: {}", event.after());
                case "u" -> log.info("Order {} status {} -> {}",
                        event.after().id(), event.before().status(), event.after().status());
                case "d" -> log.info("Order deleted: {}", event.before().id());
                default -> log.warn("Unknown op: {}", event.op());
            }
        });
    }
}

@Payload(required = false) allows Spring to pass null to the method when it encounters a tombstone. Without this annotation, Spring will report a conversion error. The "r" branch is combined with "c" because from a consumer’s perspective, a row already in the snapshot is the same as a newly created row.

Run the application with the data from section 4, and you’ll see:

INFO  OrderCdcListener : Order created: OrderRow[id=1, customerEmail=an@example.com, status=NEW, total=149.90]
INFO  OrderCdcListener : Order 1 status NEW -> PAID

6. Testing the Parser

The parsing logic is the most fragile part when Debezium or the schema changes. You can test this part without needing Kafka:

class OrderChangeParserTest {

    private final OrderChangeParser parser = new OrderChangeParser(new ObjectMapper());

    @Test
    void parsesUpdateEvent() {
        String json = """
            {"before":{"id":1,"customer_email":"an@example.com","status":"NEW","total":"149.90"},
             "after":{"id":1,"customer_email":"an@example.com","status":"PAID","total":"149.90"},
             "source":{"connector":"postgresql"},
             "op":"u","ts_ms":1760170000123}
            """;

        OrderChangeEvent event = parser.parse(json).orElseThrow();

        assertThat(event.op()).isEqualTo("u");
        assertThat(event.before().status()).isEqualTo("NEW");
        assertThat(event.after().status()).isEqualTo("PAID");
        assertThat(event.after().total()).isEqualByComparingTo("149.90");
    }

    @Test
    void returnsEmptyForTombstone() {
        assertThat(parser.parse(null)).isEmpty();
    }

    @Test
    void rejectsMalformedJson() {
        assertThatThrownBy(() -> parser.parse("{not json"))
                .isInstanceOf(IllegalArgumentException.class);
    }
}

The first test checks the mapping of the customer_email column name and BigDecimal type. The second test ensures that tombstones don’t cause the listener to fail. The third test confirms that malformed JSON is reported clearly rather than being silently swallowed.

7. Common Errors and When Not to Use CDC

Note: A replication slot keeps the WAL until Debezium confirms it has read it. If the connector stops for many days, the WAL will grow and could fill up the database’s disk. You need to monitor pg_replication_slots and delete the slot when removing the connector: SELECT pg_drop_replication_slot('shop_slot');.

Some other errors you’ll commonly encounter:

• Consumer is not idempotent. Debezium and Kafka Connect guarantee at-least-once delivery. After restart, the same event may arrive twice. Use id together with source.lsn to skip duplicates, or design your operations as upserts.
• NUMERIC values become incomprehensible base64 strings. You forgot to configure decimal.handling.mode. Similarly, TIMESTAMP columns are sent by default as microseconds since epoch, requiring manual conversion.
• before is always null. The table hasn’t been set with REPLICA IDENTITY FULL.
• Initial snapshot takes too long. With large tables, snapshot.mode=initial will read the entire table before streaming. Consider no_data if you only need changes from now on, or use incremental snapshots.
• Internal data leaks. Events reflect the exact table structure. If you pass them directly to another service, you’re exposing your database schema. For business events, combine Debezium with the Outbox pattern: your application writes to an outbox table in the same transaction, and Debezium only captures that table.
• Using the postgres account in production. Create a separate user with REPLICATION permission and SELECT permission on the tables you need to capture.

You shouldn’t use CDC when:

• You only have a single application that can emit events in its own process. In that case, a transactional outbox with a simple poller is sufficient.
• You need events with business semantics like “OrderPaid” instead of “row X changed column status”. Raw CDC only knows that data changed, not why.
• Your operations team is not ready to manage Kafka Connect, replication slots, and schema evolution.

8. Conclusion

In this article, you set up PostgreSQL with logical replication, Kafka, and Debezium, then registered a connector to capture the orders table. You also learned about the structure of change events with before, after, and op fields, then wrote a Spring Boot consumer that handles tombstones and includes tests for the parsing logic.

CDC eliminates dual-write risks because the database becomes the single source of truth. In return, you must monitor replication slots, write idempotent consumers, and consider using the Outbox pattern when you need clear business events.