Change Data Capture avec Debezium et Kafka

1. Aperçu

Quand un service a besoin de savoir que les données d’une autre base viennent de changer, l’approche habituelle est d’écrire dans la base puis d’envoyer un message sur Kafka. Cette approche s’appelle dual-write et risque facilement de causer des divergences. Change Data Capture (CDC) résout ce problème en lisant directement le log des transactions de la base.

Dans cet article, vous allez mettre en place PostgreSQL, Kafka et Debezium avec Docker Compose, puis enregistrer un connecteur. Ensuite vous écrirez un consumer Spring Boot pour traiter les événements INSERT, UPDATE et DELETE de la table orders.

2. Pourquoi CDC

Regardez ce code dual-write familier :

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

Ce code a deux faiblesses :

• Si Kafka échoue après un save réussi, la base a une commande mais aucun événement n’est envoyé.
• Si send réussit mais que la transaction fait un rollback, le consumer reçoit un événement pour une commande inexistante.

CDC inverse ce flux. L’application écrit uniquement dans la base. Un processus séparé lit le log des transactions (WAL avec PostgreSQL, binlog avec MySQL) et émet un événement pour chaque changement committé. Puisque le log ne contient que les changements committés, l’événement correspond toujours aux données réelles.

Debezium est un ensemble de connecteurs CDC qui s’exécutent sur Kafka Connect. Le flux de données est le suivant :

Application ──écrit──> PostgreSQL ──WAL──> Debezium (Kafka Connect) ──> Topic Kafka ──> Consumer

3. Mettre en place l’environnement

3.1. Docker Compose

Vous avez besoin de trois containers : PostgreSQL avec la réplication logique activée, Kafka en mode KRaft et Kafka Connect avec le plugin Debezium déjà disponible.

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

Il y a trois points à noter dans ce fichier :

• wal_level=logical est obligatoire. Sans ce paramètre, Debezium ne peut pas créer le replication slot.
• Kafka a deux listeners. Le container utilise kafka:29092, alors que l’application Spring Boot tournant sur votre machine utilise localhost:9092.
• Kafka Connect stocke la configuration, les offsets et l’état dans trois topics distincts. Grâce à cela, après un restart du connecteur, il reprendra la lecture depuis sa position précédente dans le WAL.

Démarrez et créez la 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 demande à PostgreSQL d’enregistrer la ligne complète dans le WAL. Sans cette commande, le champ before dans l’événement UPDATE serait null, et l’événement DELETE ne contiendrait que la clé primaire.

3.2. Enregistrer le connecteur

Kafka Connect expose une API REST sur le port 8083. Vous enregistrez le connecteur PostgreSQL ainsi :

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"
  }
}'

Signification des propriétés importantes :

• plugin.name: pgoutput utilise le plugin de décodage intégré de PostgreSQL 10+, vous n’avez donc pas à installer d’extension supplémentaire.
• topic.prefix combiné avec le schéma et le nom de la table génère le nom du topic. Ici, le topic est shop.public.orders.
• table.include.list limite les tables à capturer.
• decimal.handling.mode: string fait que les colonnes NUMERIC sont envoyées sous forme de chaînes "149.90". Par défaut, Debezium encode cette valeur en bytes base64, ce qui est très difficile à lire.
• schemas.enable: false supprime la partie schema accompagnant chaque message JSON, rendant le payload plus compact.

Vérifiez le statut du connecteur :

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. Structure de l’événement de changement

Ajoutez puis mettez à jour une commande :

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;"

Puis lisez le topic avec le console consumer intégré dans l’image Kafka :

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

L’événement UPDATE ressemble à ceci (champs source résumés) :

{
  "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
}

Le champ op indique le type de changement :

op	Signification	before	after
c	INSERT	null	nouvelle ligne
u	UPDATE	ancienne ligne	nouvelle ligne
d	DELETE	ancienne ligne	null
r	Lecture dans le snapshot initial	null	ligne existante

Quand une ligne est supprimée, Debezium envoie un tombstone supplémentaire, c’est-à-dire un message ayant la même clé mais une valeur null. Le tombstone permet au log compaction de Kafka de nettoyer complètement cette clé. Votre consumer doit gérer une valeur null.

5. Consommer les événements avec Spring Boot

5.1. Dépendances et 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>

Les versions de ces dépendances sont gérées par 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

Ici le consumer reçoit la value sous forme de String puis l’analyse lui-même. Cette approche vous donne plus de contrôle sur les tombstones et les erreurs JSON comparé à l’utilisation de JsonDeserializer.

5.2. Modèle et 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) ignore le champ source et tout nouveau champ que Debezium pourrait ajouter plus tard. Jackson convertit automatiquement la chaîne "149.90" en BigDecimal, vous ne perdez donc pas de précision comme avec 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);
        }
    }
}

Le parser retourne Optional.empty() pour un tombstone, donc le listener peut simplement ignorer ce cas.

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) permet à Spring de passer null à la méthode lors d’une rencontre avec un tombstone. Sans cette annotation, Spring signalera une erreur de conversion. Le cas "r" est fusionné avec "c" car pour un consumer, une ligne existant déjà dans le snapshot revient au même qu’une ligne juste créée.

En exécutant l’application avec les données de la section 4, vous verrez :

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

6. Tester le parser

La logique de parsing est la partie la plus fragile quand Debezium ou le schéma change. Vous pouvez tester cette partie sans 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);
    }
}

Le premier test vérifie le mappage du nom de colonne customer_email et du type BigDecimal. Le deuxième test assure qu’un tombstone ne fait pas échouer le listener. Le troisième test confirme que du JSON mal formé génère une erreur claire au lieu d’être silencieusement englouti.

7. Erreurs courantes et quand ne pas utiliser CDC

Attention : Le replication slot garde le WAL jusqu’à ce que Debezium confirme l’avoir lu. Si le connecteur s’arrête pendant plusieurs jours, le WAL gonflera et pourrait remplir le disque de la base. Vous devez surveiller pg_replication_slots et supprimer le slot quand vous retirez le connecteur : SELECT pg_drop_replication_slot('shop_slot');.

Voici quelques autres erreurs que vous rencontrerez souvent :

• Consumer non idempotent. Debezium et Kafka Connect garantissent at-least-once. Après un restart, le même événement peut arriver deux fois. Utilisez id combiné avec source.lsn pour ignorer les doublons, ou concevez les opérations comme un upsert.
• Valeur NUMERIC en base64 incompréhensible. Vous avez oublié de configurer decimal.handling.mode. De même, les colonnes TIMESTAMP sont par défaut envoyées en nombre de microsecondes depuis l’époque, vous devez donc convertir vous-même.
• before toujours null. La table n’a pas REPLICA IDENTITY FULL.
• Snapshot initial très long. Avec une grande table, snapshot.mode=initial lira la table entière avant de faire du streaming. Envisagez no_data si vous ne voulez que les changements à partir de maintenant, ou utilisez incremental snapshot.
• Fuite de données internes. L’événement reflète la structure exacte de la table. Si vous le passez directement à un autre service, vous divulguez votre schéma de base. Pour les événements métier, combinez Debezium avec le Outbox pattern : l’application écrit dans une table outbox dans la même transaction, et Debezium ne capture que cette table.
• Utiliser le compte postgres en production. Créez un utilisateur dédié avec les droits REPLICATION et SELECT sur les tables à capturer.

Vous ne devriez pas utiliser CDC quand :

• Vous n’avez qu’une seule application et elle peut produire ses propres événements dans le même processus. Dans ce cas, un transactional outbox avec un simple poller suffit.
• Vous avez besoin d’événements portant une signification métier comme « OrderPaid » plutôt que « la ligne X a changé la colonne status ». CDC brut ne sait que la donnée a changé, pas pourquoi.
• Votre équipe opérationnelle n’est pas prête à gérer Kafka Connect, les replication slots et l’évolution du schéma.

8. Conclusion

Dans cet article, vous avez mis en place PostgreSQL avec la réplication logique, Kafka et Debezium, puis enregistré un connecteur pour capturer la table orders. Vous avez aussi découvert la structure des événements de changement avec les champs before, after et op, puis écrit un consumer Spring Boot gérant les tombstones, avec des tests pour la partie parsing.

CDC élimine le risque du dual-write car la base devient la seule source de vérité. En contrepartie, vous devez surveiller les replication slots, écrire des consumers idempotents et envisager le Outbox pattern quand vous avez besoin d’événements métier clairs.