Kafka Connect (2/4) — Debezium Postgres CDC

🎯 OBJECTIF

Comprendre comment :

Le WAL Postgres, LSN et logical decoding fonctionnent — les couches que Debezium exploite
Un replication slot est une ligne inerte dans pg_replication_slots et son rôle de marqueur de rétention WAL
Slot vs wal_sender — deux choses distinctes à ne pas confondre
Configurer Postgres (postgresql.conf, user, publication) et Debezium (config JSON complète)
Interpréter chaque champ de la config Debezium en contexte de production

🧠 MODÈLE MENTAL

Debezium fait du CDC (Change Data Capture) en lisant le journal de réplication Postgres, pas en faisant du polling SQL. Il se comporte comme une standby logique : Postgres lui streame les événements du WAL décodés en opérations logiques (INSERT/UPDATE/DELETE). Debezium les convertit en events Kafka.

L'analogie qui aide : le WAL est un journal de bord séquentiel. Debezium est un lecteur de journal avec un signet (le replication slot). Le slot dit à Postgres "ne jette rien avant ma page marquée". Tant que le slot existe et que le signet n'avance pas, Postgres accumule les pages. C'est à la fois la garantie de ne rien rater et la bombe à retardement si le lecteur s'arrête.

Prérequis : kafka-connect-1-fondamentaux (workers, connectors, tasks, topics internes).

1📖 Glossaire Postgres CDC

WAL (Write-Ahead Log) — journal append-only dans pg_wal/ où Postgres écrit chaque modification avant de l'appliquer à la table. Segments de 16 MB.
LSN (Log Sequence Number) — position dans le WAL. Format 0/16A3D8. Strictement croissant, soustrayable.
Logical decoding — capacité à décoder le WAL physique (octets de pages) en événements logiques (INSERT dans table T, valeurs (c1, c2)). Requiert wal_level=logical.
Output plugin — module qui formate les événements décodés. pgoutput (natif) ou wal2json (extension).
Replication slot — ligne dans pg_replication_slots qui indique où un consommateur en est dans le WAL. Marqueur de rétention.
wal_sender — processus éphémère côté Postgres qui lit le WAL, décode via le plugin, streame sur TCP. Un par connexion active.
Publication — objet SQL qui liste les tables à répliquer. Requis avec pgoutput.
REPLICA IDENTITY — attribut de table qui définit les colonnes incluses dans les événements UPDATE/DELETE du WAL logique.
confirmed_flush_lsn — LSN jusqu'où le consommateur a confirmé avoir lu. Le WAL avant cette position peut être recyclé.

2Le WAL et les 3 couches de CDC

Debezium s'appuie sur trois couches Postgres empilées.

flowchart LR
  subgraph WAL["Couche 1 — WAL"]
    W["pg_wal/\nappend-only\nLSN croissant"]
  end
  subgraph DECODE["Couche 2 — Logical decoding"]
    P["pgoutput plugin\n(natif Postgres 10+)"]
  end
  subgraph FILTER["Couche 3 — Filtrage"]
    PUB["Publication SQL\n(tables capturées)"]
    SLOT["Replication slot\n(marqueur rétention)"]
    RI["REPLICA IDENTITY\n(colonnes dans events)"]
  end
  WAL --> DECODE --> FILTER

mermaid

Couche 1 — WAL : journal séquentiel de toutes les modifications. wal_level=logical ajoute les informations nécessaires au décodage logique (~10-20% de WAL en plus).

Couche 2 — Logical decoding : traduit le WAL physique en événements logiques. pgoutput est natif depuis Postgres 10 — pas d'installation, performant, recommandé par défaut. wal2json est une extension JSON plus lente mais lisible manuellement.

Couche 3 — Filtrage : la publication liste les tables, le slot marque la position, REPLICA IDENTITY définit ce qu'on voit dans les events.

3Le replication slot — la mécanique exacte

Réflexe trompeur : imaginer qu'un slot extrait du WAL, le copie, ou calcule des bytes. Il ne fait rien de tout ça.

Un slot = une ligne dans pg_replication_slots, quelques kilobytes sur disque. Inerte — pas de processus, pas de thread, pas de logique propre.

SELECT slot_name, active, active_pid,
       restart_lsn, confirmed_flush_lsn
FROM pg_replication_slots
WHERE slot_name = 'sie_debezium_slot';

sql

Colonne	Contenu
`active`	Un client est-il connecté maintenant ?
`active_pid`	PID du wal_sender actif (si connecté)
`restart_lsn`	Où Postgres peut recommencer la lecture en cas de reconnexion
`confirmed_flush_lsn`	Jusqu'où Debezium a confirmé avoir traité

La règle de rétention du checkpointer : tant que le slot existe, Postgres refuse de recycler tout segment WAL contenant des LSN ≥ restart_lsn. C'est là que vit tout le pouvoir — et tout le danger.

min_lsn_retenu = min(restart_lsn de tous les slots)
pour chaque segment WAL :
    si segment.max_lsn < min_lsn_retenu → recyclable
    sinon → Postgres garde

Slot vs wal_sender — deux choses distinctes

	Slot	wal_sender
Nature	Ligne persistée dans le catalogue	Processus éphémère côté Postgres
Durée de vie	Jusqu'à `DROP` explicite	Durée de la connexion
Rôle	Marqueur de rétention WAL	Lit, décode, stream au client
Limité par	`max_replication_slots`	`max_wal_senders`

Le slot survit à la déconnexion de Debezium — c'est ce qui permet la reprise transparente après crash. Mais un slot orphelin (Debezium supprimé sans drop du slot) retient le WAL indéfiniment. Voir kafka-connect-3-slot-lag-monitoring pour la gestion des slots orphelins.

4Prérequis Postgres

postgresql.conf

wal_level = logical          # requiert redémarrage Postgres
max_wal_senders = 10         # 1 par connexion Debezium + standbys + backups
max_replication_slots = 10   # 1 par connector Debezium + standbys

Ne pas confondre tasks.max (Connect) et max_replication_slots (Postgres)

tasks.max=1 est le plafond Connect pour Debezium PG. max_replication_slots dimensionne combien de slots l'instance Postgres entière peut héberger — pour tous les consommateurs : Debezium, standbys, pg_basebackup.

SQL de préparation

-- User dédié avec droit de réplication
CREATE ROLE debezium WITH REPLICATION LOGIN PASSWORD '...';
GRANT CONNECT ON DATABASE sie TO debezium;
GRANT USAGE ON SCHEMA stock TO debezium;
GRANT SELECT ON ALL TABLES IN SCHEMA stock TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA stock GRANT SELECT ON TABLES TO debezium;

-- Publication (filtre côté DB)
CREATE PUBLICATION sie_debezium_pub
  FOR TABLE stock.stock, stock.stock_movement;

-- Table sans PK → REPLICA IDENTITY FULL obligatoire
ALTER TABLE stock.ajustement REPLICA IDENTITY FULL;

sql

REPLICA IDENTITY : par défaut (DEFAULT), seules les colonnes de la PK sont incluses dans les events UPDATE/DELETE. Sans PK → les updates/deletes ne portent rien d'exploitable. Solution : REPLICA IDENTITY FULL (toutes les colonnes, plus coûteux).

5Config Debezium complète — champ par champ

{
  "name": "sie-stock-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",

    "database.hostname": "pg-sie.prod.internal",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/secrets/db-creds:pg_debezium_pwd}",
    "database.dbname": "sie",

    "plugin.name": "pgoutput",
    "slot.name": "sie_debezium_slot",
    "publication.name": "sie_debezium_pub",
    "publication.autocreate.mode": "disabled",

    "topic.prefix": "sie",
    "table.include.list": "stock.stock,stock.stock_movement",

    "snapshot.mode": "initial",

    "heartbeat.interval.ms": "10000",
    "heartbeat.action.query": "UPDATE public.debezium_heartbeat SET ts = now() WHERE id = 1",

    "key.converter": "io.confluent.connect.avro.AvroConverter",
    "key.converter.schema.registry.url": "http://schema-registry:8081",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",

    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite"
  }
}

json

Champ	Valeur recommandée	Rôle
`tasks.max`	`1`	WAL séquentiel — un seul curseur. Valeur supérieure sans effet.
`database.password`	`${file:/secrets/...}`	FileConfigProvider — évite le mot de passe en clair dans `connect-configs`.
`plugin.name`	`pgoutput`	Natif, plus performant que `wal2json`. Toujours `pgoutput` sauf compat historique.
`slot.name`	nom stable	Créé au 1er démarrage. Ne jamais changer en prod — perte de la position LSN.
`publication.autocreate.mode`	`disabled`	En prod, la publication est gérée via SQL explicite / Flyway.
`topic.prefix`	`sie`	Topics produits : `{prefix}.{schema}.{table}` → `sie.stock.stock`.
`snapshot.mode`	`initial`	Au 1er démarrage, SELECT * sur chaque table puis streaming WAL. `never` = pas de snapshot (trou historique). `always` = snapshot à chaque démarrage.
`heartbeat.interval.ms`	`10000`	Critique si tables capturées peu actives — voir kafka-connect-3-slot-lag-monitoring.
`transforms.unwrap.drop.tombstones`	`false`	Garde les tombstones (value=null sur DELETE) pour propager vers le sink.
`transforms.unwrap.delete.handling.mode`	`rewrite`	DELETE produit message `{id, __deleted=true}` + tombstone. `drop` = perte silencieuse des DELETE.

⚡ TL;DR — chaque concept en une ligne

WAL + Logical decoding ✓ Journal append-only décodé en événements logiques — c'est la source que Debezium consomme. ⚠ wal_level=logical est requis et demande un redémarrage Postgres — à planifier en avance.

Replication slot ✓ Ligne inerte dans pg_replication_slots qui retient le WAL tant que Debezium n'a pas confirmé sa lecture. ⚠ Un slot orphelin (Debezium détruit sans DROP du slot) retient le WAL indéfiniment → disque plein.

pgoutput vs wal2json ✓ pgoutput est natif, binaire, plus performant — recommandé par défaut. ⚠ wal2json reste utile pour débugger manuellement via pg_recvlogical, mais plus coûteux en prod.

REPLICA IDENTITY ✓ FULL capture toutes les colonnes dans les events UPDATE/DELETE — indispensable si pas de PK. ⚠ DEFAULT (PK seule) rend les UPDATE/DELETE inutilisables sur les tables sans clé primaire.

🎓 À retenir

Slot ≠ wal_sender — le slot est une ligne persistante qui survit à la déconnexion. Le wal_sender est le processus éphémère qui stream pendant la connexion. Supprimer Debezium ne détruit pas le slot — il faut le faire explicitement.
slot.name ne doit jamais changer en prod — un rename crée un nouveau slot, Postgres lui attribue le LSN courant, l'historique WAL entre l'ancien et le nouveau est perdu. Debezium démarrera un nouveau snapshot.
heartbeat.action.query doit écrire dans une table de la publication — un INSERT dans une table non publiée ne génère aucun event que Debezium peut traiter, le curseur ne bougera pas.

Sources

Debezium — PostgreSQL Connector Documentation — config complète, WAL, snapshot modes, REPLICA IDENTITY
PostgreSQL — Logical Replication — publications, slots, wal_level
PostgreSQL — pg_replication_slots — colonnes restart_lsn, confirmed_flush_lsn
kafka-connect-1-fondamentaux — architecture Connect, workers, topics internes
kafka-connect-3-slot-lag-monitoring — slot lag, modes de défaillance, procédures de reset
kafka-connect-4-sink-smt-dlq — JDBC Sink qui consomme les topics Debezium