Pipeline CDC PostgreSQL avec Debezium 3.4 et Kafka Connect

Le Change Data Capture (CDC) consiste à transformer en flux d’événements toutes les modifications qui se produisent dans une base de données : un INSERT, un UPDATE, un DELETE deviennent autant de messages publiés sur un topic Kafka. Sans toucher au code applicatif, sans trigger SQL, sans batch nocturne. Debezium est le projet open-source de référence pour ce besoin, maintenu par Red Hat sous licence Apache 2.0. Sa version stable au moment d’écrire ces lignes est Debezium 3.4.0.Final, sortie le 16 décembre 2025, et elle s’intègre nativement à Kafka Connect, le runtime d’intégration fourni avec Apache Kafka 4.2.

Ce tutoriel construit un pipeline CDC complet de PostgreSQL vers Kafka : on configure PostgreSQL pour exposer son journal WAL, on déploie Kafka Connect en distribué, on installe le connecteur Debezium PostgreSQL, et on observe les événements de changement arriver en quasi temps réel sur un topic Kafka.

Prérequis

• Un cluster Kafka 4.2 fonctionnel — voir le tutoriel de mise en route KRaft
• Une instance PostgreSQL 14, 15, 16, 17 ou 18 (PostgreSQL 13 a été retiré du périmètre de test de Debezium 3.4)
• Docker 27+ pour déployer Kafka Connect et le connecteur
• Connaissances de base en SQL et JSON
• Temps estimé : 90 minutes

Étape 1 — Comprendre pourquoi CDC et pourquoi Debezium

Il existe historiquement trois grandes manières d’extraire les changements d’une base. La méthode par polling exécute une requête SELECT régulière sur une colonne updated_at ; simple, mais consommatrice en charge et incapable de capturer les DELETE. La méthode par trigger SQL écrit chaque changement dans une table d’audit ; complète mais intrusive et coûteuse en performances d’écriture. La méthode par log replication lit directement le journal binaire que la base utilise pour la réplication ; aucune modification du schéma, aucun impact sur les performances de l’application, et capture exacte de chaque DML.

Debezium implémente la troisième approche. Pour PostgreSQL, il consomme le logical replication slot via le plugin pgoutput intégré nativement depuis PostgreSQL 10. Pour MySQL, il lit le binlog en format ROW. Pour MongoDB, il s’abonne à l’oplog. Pour SQL Server, il utilise le mécanisme CDC natif. Le résultat est uniforme : un flux d’événements JSON ou Avro publié sur un topic Kafka, qu’importe la source.

Étape 2 — Préparer PostgreSQL pour la réplication logique

PostgreSQL nécessite trois ajustements pour exposer son journal en mode logique. Sur Ubuntu, le fichier de configuration se trouve typiquement à /etc/postgresql/16/main/postgresql.conf.

sudo nano /etc/postgresql/16/main/postgresql.conf
# Ajouter ou modifier
wal_level = logical
max_wal_senders = 10
max_replication_slots = 10

wal_level=logical est le paramètre central — il indique au moteur de stocker dans le WAL les informations sémantiques nécessaires au décodage logique. max_wal_senders détermine combien de processus de réplication peuvent tourner en parallèle ; un par connecteur Debezium plus une marge. max_replication_slots joue le même rôle pour les slots logiques eux-mêmes. Après la modification, un redémarrage PostgreSQL est obligatoire : sudo systemctl restart postgresql.

On crée ensuite l’utilisateur de réplication dédié à Debezium et on lui accorde les droits minimums nécessaires :

CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD 'changez-moi';
GRANT CONNECT ON DATABASE itskc TO debezium;
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;

-- Table d'exemple
CREATE TABLE commandes (
  id BIGSERIAL PRIMARY KEY,
  client_telephone TEXT NOT NULL,
  montant_fcfa INTEGER NOT NULL,
  statut TEXT NOT NULL DEFAULT 'EN_ATTENTE',
  cree_le TIMESTAMPTZ DEFAULT now()
);

Le rôle REPLICATION est le seul droit non-trivial : c’est lui qui autorise la lecture du WAL via le protocole de streaming. Le mot de passe sera à passer en clair au connecteur, donc on l’isole sur un secret Docker ou Kubernetes en production. Le SELECT sur les tables est nécessaire pour le snapshot initial — Debezium copie l’état présent avant de commencer à lire le WAL.

Dernière action côté base : autoriser la connexion de l’utilisateur de réplication dans pg_hba.conf.

# /etc/postgresql/16/main/pg_hba.conf
host    replication     debezium        10.0.0.0/24             scram-sha-256
host    itskc           debezium        10.0.0.0/24             scram-sha-256

Le sous-réseau 10.0.0.0/24 est à remplacer par le CIDR réel où tourne Kafka Connect. sudo systemctl reload postgresql applique le changement sans redémarrage complet.

Étape 3 — Déployer Kafka Connect en mode distribué

Kafka Connect est un runtime qui exécute des connecteurs source (qui injectent des données dans Kafka) et des connecteurs sink (qui extraient les données vers d’autres systèmes). En mode distribué, plusieurs workers se partagent automatiquement les tâches via leurs propres topics internes Kafka — c’est ce qu’on utilise en production.

# docker-compose.yml
services:
  kafka-connect:
    image: debezium/connect:3.4.0.Final
    container_name: connect
    restart: unless-stopped
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka1.example.com:9092,kafka2.example.com:9092
      GROUP_ID: connect-cluster-1
      CONFIG_STORAGE_TOPIC: connect_configs
      OFFSET_STORAGE_TOPIC: connect_offsets
      STATUS_STORAGE_TOPIC: connect_status
      CONFIG_STORAGE_REPLICATION_FACTOR: 3
      OFFSET_STORAGE_REPLICATION_FACTOR: 3
      STATUS_STORAGE_REPLICATION_FACTOR: 3
      KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
      KEY_CONVERTER_SCHEMAS_ENABLE: "false"
      VALUE_CONVERTER_SCHEMAS_ENABLE: "false"

docker compose up -d kafka-connect
curl -s http://localhost:8083/ | jq
# Sortie : {"version":"4.2.0", "commit":"...", "kafka_cluster_id":"WJxR2KdLT_qV9bGq3oZc4w"}
curl -s http://localhost:8083/connector-plugins | jq '.[].class' | grep -i postgres
# Sortie attendue : "io.debezium.connector.postgresql.PostgresConnector"

L’image debezium/connect:3.4.0.Final embarque déjà tous les connecteurs Debezium précompilés. Le port 8083 sert l’API REST de Kafka Connect, par laquelle on installera et inspectera les connecteurs. Les trois topics connect_configs, connect_offsets et connect_status sont créés automatiquement au premier démarrage et stockent l’état persistent du cluster Connect. Un facteur de réplication de 3 est fortement recommandé en production pour éviter de perdre les configurations en cas de panne de broker.

Étape 4 — Créer le connecteur Debezium PostgreSQL

La création d’un connecteur se fait par un POST JSON sur l’API REST de Kafka Connect. Le payload ci-dessous instancie le connecteur Debezium PostgreSQL en mode pgoutput sur la table commandes.

cat > commandes-connector.json <<'EOF'
{
  "name": "commandes-pg-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres.example.com",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "changez-moi",
    "database.dbname": "itskc",
    "topic.prefix": "cdc.itskc",
    "table.include.list": "public.commandes",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_commandes",
    "publication.name": "debezium_pub_commandes",
    "publication.autocreate.mode": "filtered",
    "snapshot.mode": "initial",
    "tombstones.on.delete": "true",
    "decimal.handling.mode": "string",
    "time.precision.mode": "connect"
  }
}
EOF

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

Trois paramètres essentiels. topic.prefix sert de namespace : les topics créés s’appellent cdc.itskc.public.commandes. plugin.name=pgoutput indique l’utilisation du plugin natif PostgreSQL — pas besoin d’installer wal2json ou autre extension. publication.autocreate.mode=filtered demande à Debezium de créer automatiquement la publication PostgreSQL avec uniquement les tables listées (préféré à all_tables pour des raisons de sécurité). La réponse 201 indique le succès ; curl http://localhost:8083/connectors/commandes-pg-source/status vérifie l’état du connecteur et de sa tâche.

Étape 5 — Observer le premier événement de changement

Avec un consumer Kafka CLI ouvert sur le topic cdc.itskc.public.commandes, on insère une ligne en base et on regarde l’événement apparaître.

# Terminal 1 — consumer
/opt/kafka/bin/kafka-console-consumer.sh \
  --bootstrap-server kafka1.example.com:9092 \
  --topic cdc.itskc.public.commandes \
  --from-beginning \
  --property print.key=true | jq

# Terminal 2 — psql
psql -h postgres.example.com -U itskc -d itskc
INSERT INTO commandes (client_telephone, montant_fcfa)
  VALUES ('+221771234567', 25000);

L’événement reçu côté Kafka a une structure caractéristique Debezium : un objet avec les champs before, after, source et op. op="c" indique une création, "u" une mise à jour, "d" une suppression, "r" un événement issu du snapshot initial. before est null pour un INSERT, contient l’état précédent pour un UPDATE. source liste la base, le schéma, la table, le LSN PostgreSQL et l’horodatage du commit — utile pour les requêtes en temps réel et la traçabilité forensique.

Un signal de réussite immédiat : un UPDATE sur la table doit générer un événement avec before non null contenant les anciennes valeurs et after avec les nouvelles. Si before reste null sur un UPDATE, la replica identity de la table est probablement à DEFAULT et ne contient que la clé primaire ; pour avoir les valeurs avant, exécuter ALTER TABLE commandes REPLICA IDENTITY FULL.

Étape 6 — Brancher un sink vers Meilisearch ou Postgres

Le pipeline n’est complet qu’avec un consommateur en aval. Le pattern le plus commun est un connecteur sink qui copie les événements vers un index de recherche ou une base analytique. Pour Meilisearch ou OpenSearch, le connecteur kafka-connect-elasticsearch publié par Aiven sous licence Apache 2.0 fait le travail.

{
  "name": "commandes-meili-sink",
  "config": {
    "connector.class": "io.aiven.kafka.connect.opensearch.OpenSearchSinkConnector",
    "topics": "cdc.itskc.public.commandes",
    "connection.url": "http://meilisearch.example.com:7700",
    "tasks.max": "2",
    "key.ignore": "false",
    "schema.ignore": "true",
    "behavior.on.null.values": "delete"
  }
}

Le paramètre behavior.on.null.values=delete exploite la sémantique des tombstones Debezium : quand une ligne est supprimée en PostgreSQL, Debezium publie un message avec value=null (le tombstone), ce qui supprime automatiquement le document correspondant dans l’index. Pour Meilisearch spécifiquement, voir l’article Meilisearch + Drizzle ORM : indexation auto Postgres → Meilisearch qui détaille une autre approche par triggers logiques.

Étape 7 — Vérifier la santé du pipeline

En production, on monitore quatre métriques pour s’assurer que le pipeline est sain. Le lag du slot de réplication PostgreSQL — calculé par SELECT pg_wal_lsn_diff(pg_current_wal_lsn(), confirmed_flush_lsn) FROM pg_replication_slots WHERE slot_name='debezium_commandes' — donne en octets le retard du connecteur. Une valeur stable proche de zéro est saine ; une croissance continue indique que Debezium ne suit plus la cadence.

curl -s http://localhost:8083/connectors/commandes-pg-source/status | jq
# Sortie attendue
# { "name": "...", "connector": { "state": "RUNNING" }, "tasks": [ { "id": 0, "state": "RUNNING" } ] }

L’état RUNNING sur le connecteur et sur sa tâche est le signal nominal. Un FAILED est presque toujours dû à un changement de schéma incompatible ou à une perte de connexion à PostgreSQL trop longue — la cause précise est dans le champ trace de la réponse.

Erreurs fréquentes

Considérations sécurité et conformité

Le CDC capture par construction toutes les modifications, y compris les champs sensibles. Pour les plateformes ouest-africaines qui manipulent des numéros de téléphone, des numéros d’identité nationale ou des montants de transactions, deux protections sont essentielles. La première : utiliser Single Message Transforms de Kafka Connect pour masquer ou hasher des champs avant publication — le transform MaskField remplace la valeur d’un champ par des étoiles, HoistField et ExtractField permettent de restructurer le payload pour exclure des données. La seconde : chiffrer les topics CDC au repos via la fonction field-level encryption de Confluent Platform ou via un wrapper applicatif côté consumer.

La gouvernance doit aussi inclure une procédure de rétention. Les topics CDC ne devraient pas être conservés au-delà de la durée légale applicable — la loi 2008-12 sénégalaise sur les données personnelles fixe la durée à ce qui est strictement nécessaire à la finalité. Pour des événements opérationnels, sept à trente jours suffisent ; au-delà, on archive sélectivement vers un object store chiffré.

Coût opérationnel

Un connecteur Debezium PostgreSQL consomme typiquement entre 200 et 500 Mo de RAM pour suivre une charge de quelques milliers d’événements par seconde. Sur Kafka Connect, on dimensionne en règle générale à un worker par CPU disponible, plus une marge de 30 % pour les pics. Pour une PME ivoirienne avec une à deux bases PostgreSQL d’environ 100 Go et un débit de quelques centaines d’écritures par seconde, un seul worker Kafka Connect sur 3 vCPU et 4 Go de RAM (un Hetzner CPX21 à environ 12 euros par mois post-1er avril 2026) est largement suffisant. La charge sur PostgreSQL lui-même augmente d’environ 2 à 5 % en CPU, principalement due à l’écriture du WAL en mode logique — totalement absorbable sur tout serveur sain.

Ressources et références

• Documentation Debezium PostgreSQL connector — version 3.4
• Annonce Debezium 3.4.0.Final (16 décembre 2025)
• Documentation Kafka Connect 4.2
• Documentation PostgreSQL — réplication logique
• Alternative ingestion data : Airbyte self-hosted

Single Message Transforms : enrichir, filtrer, anonymiser

Kafka Connect propose un mécanisme léger appelé Single Message Transforms (SMT) qui modifie chaque événement entre la lecture par le connecteur source et l’écriture dans le topic. Pour Debezium, c’est l’endroit où l’on peut aplatir la structure imbriquée before/after en un objet plat, ajouter le nom de l’environnement, ou supprimer des champs sensibles.

{
  "transforms": "unwrap,addEnv,mask",
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.delete.tombstone.handling.mode": "rewrite",
  "transforms.addEnv.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addEnv.static.field": "env",
  "transforms.addEnv.static.value": "production",
  "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.mask.fields": "client_telephone"
}

Le transform ExtractNewRecordState est spécifique à Debezium et transforme l’enveloppe {before, after, op, source} en un simple objet plat équivalent au contenu de after — beaucoup plus simple à consommer côté aval. Combiné à MaskField qui remplace le numéro de téléphone par des étoiles, on obtient un flux CDC déjà conforme aux exigences de minimisation des données personnelles, sans aucun code applicatif.

Snapshot modes et reprise après incident

Debezium propose plusieurs modes de snapshot. initial (défaut) copie l’état présent puis enchaîne sur le streaming WAL — idéal pour un démarrage à froid. no_data ne capture que la structure (schémas/DDL) sans émettre d’événements READ pour les lignes existantes — utile quand les données sont déjà répliquées par ailleurs et qu’on ne veut que le stream futur (remplace l’ancien never aujourd’hui déprécié). initial_only fait un snapshot puis s’arrête — utile pour exporter ponctuellement une base. when_needed n’exécute le snapshot que si Debezium détecte qu’il a perdu sa position dans le WAL. Deux modes avancés existent également : configuration_based pour piloter les sous-comportements via des propriétés dédiées, et custom pour brancher une classe Java implémentant le SPI Snapshotter.

En cas d’incident grave — par exemple le slot de réplication a été supprimé manuellement et le LSN cherché n’est plus dans le WAL — la procédure de récupération est : supprimer le connecteur, supprimer les offsets stockés dans connect_offsets pour ce connecteur, recréer le connecteur en snapshot.mode=initial. C’est pénible mais sans risque : les consumers en aval sont protégés par l’idempotence métier (clé primaire) et ne subiront que des retraitements.

Topology multi-bases et topic routing

Quand on a plusieurs bases PostgreSQL avec des tables qui poussent vers les mêmes topics applicatifs, le topic routing SMT devient indispensable. Le transform RegexRouter permet de réécrire dynamiquement le nom du topic de destination en fonction du nom de la table source. Par exemple, on peut diriger tous les événements de tables commandes_* de plusieurs schémas vers un unique topic commandes, ce qui simplifie l’architecture en aval.

{
  "transforms": "routes",
  "transforms.routes.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.routes.regex": "cdc\\.itskc\\.public\\.commandes_(.+)",
  "transforms.routes.replacement": "commandes_consolide"
}

Cette regex transforme cdc.itskc.public.commandes_dakar, cdc.itskc.public.commandes_abidjan et toute autre variante en un seul topic commandes_consolide. Tous les événements convergent vers la même partition selon la clé de message — typiquement la clé primaire — ce qui permet aux consumers de traiter linéairement l’historique consolidé.

Workflow type pour un nouveau pipeline

L’expérience montre que les pipelines CDC qui survivent à un an de production suivent à peu près tous le même rituel de mise en place. On commence par déployer le connecteur en environnement de pré-production sur une copie restaurée de la base de production. On valide pendant deux à quatre semaines la stabilité du lag, la conformité des événements, et la robustesse face aux pics d’écriture. On documente alors le runbook opérationnel : commande exacte de redémarrage du connecteur, procédure de purge de slot, contact responsable de la base. Ce n’est qu’après cette phase de validation qu’on bascule sur la base de production réelle, généralement à une heure creuse, et qu’on monitore activement les 48 premières heures. Cette discipline évite la grande majorité des incidents que connaissent les équipes qui partent directement en production.

Retour au guide principal : Apache Kafka en production en 2026 : panorama complet.