07 - Kafka Connect: Source/Sink Connector, Debezium CDC e Integrazione DB

Ogni modifica a una tabella deve arrivare su Kafka, senza scrivere un producer

Il database degli ordini cambia di continuo: righe nuove, stati aggiornati, cancellazioni. Altri servizi hanno bisogno di sapere cosa è cambiato, ma interrogare la tabella a intervalli (il polling) ha due difetti noti: costa letture inutili e non vede mai una cancellazione, perché una riga cancellata non si può più leggere. Scrivere a mano il producer che copia le modifiche significa risolvere da soli problemi già risolti: riavvio senza perdere la posizione, ripartizione del lavoro, gestione degli errori.

Kafka Connect è il runtime che risolve questa parte: tu configuri un connector con un file JSON, lui lo esegue, ricorda dove era arrivato e lo ripartisce su più processi. Per leggere le modifiche di PostgreSQL esiste Debezium, che non interroga le tabelle ma legge il log delle transazioni del database. In questo articolo vedi tre cose con codice eseguito, usando le classi vere di Kafka Connect 4.4.0 (connect-api, connect-transforms, Maven Central, SHA-1 verificati): come un task ricorda la propria posizione, come funziona una catena di trasformazioni e che cosa succede dopo un crash. Debezium e PostgreSQL non sono stati eseguiti in questo ambiente: i file di configurazione e il SQL sono schemi da adattare, verificati contro la documentazione di Debezium (versione «stable») e di Kafka 4.4, consultate il 2026-11-12.

Worker, connector e task: chi fa che cosa

Un worker è un processo JVM che esegue connector. Il connector è la definizione dell'integrazione («copia la tabella orders su Kafka») e decide in quanti task dividerla; il task è l'unità che lavora davvero. Il numero massimo lo dà tasks.max (default 1), con la clausola della documentazione che il connector può creare meno task se non riesce a raggiungere quel parallelismo: il connector Debezium per PostgreSQL, dice la sua documentazione, «always uses a single task» e ignora quel valore, perché il log delle transazioni è un flusso unico. Il formato dei byte scritti su Kafka non lo decide il connector ma i converter (key.converter, value.converter), così che qualunque connector possa lavorare con qualunque formato.

Le due modalità di esecuzione, secondo la guida utente, sono standalone (un processo solo, adatto per esempio alla raccolta di file di log, ma senza tolleranza ai guasti) e distributed, dove più worker formano un cluster che ripartisce il lavoro automaticamente. In modalità distribuita configurazioni, offset e stati dei task vivono in tre topic Kafka (config.storage.topic, offset.storage.topic, status.storage.topic), che la documentazione consiglia di creare a mano. I connector si gestiscono con la REST API del worker: elenco con GET /connectors, creazione con POST /connectors, stato con GET /connectors/{nome}/status, e inoltre pause, resume e restart.

Offset: come un connector ricorda dove era arrivato

Ogni record che un source connector produce porta con sé due piccole mappe: una source partition (di che sorgente si tratta, per esempio la tabella) e un source offset (fino a dove si è letto). Connect salva periodicamente l'ultimo offset nel topic degli offset e lo restituisce al task quando riparte. Il listato usa le classi vere SourceTask e SourceRecord dell'API di Connect e simula soltanto il lavoro del worker, che qui è un piccolo ciclo in memoria: legge un lotto, lo scrive sul «topic», e solo dopo salva l'offset.

import org.apache.kafka.connect.source.*;
import org.apache.kafka.connect.storage.OffsetStorageReader;
import java.util.*;

public class OffsetSorgente {
    // Un task sorgente "da manuale": legge righe numerate e dichiara a Connect dove è arrivato.
    static class TabellaTask extends SourceTask {
        static final Map<String, String> PARTIZIONE = Map.of("tabella", "orders");
        long prossimoId = 1;
        @Override public String version() { return "demo"; }
        @Override public void start(Map<String, String> props) {
            Map<String, Object> salvato = context.offsetStorageReader().offset(PARTIZIONE);
            if (salvato != null) prossimoId = (Long) salvato.get("prossimo");      // riparte da dove era arrivato
        }
        @Override public List<SourceRecord> poll() {
            List<SourceRecord> batch = new ArrayList<>();
            for (int i = 0; i < 2 && prossimoId <= 6; i++, prossimoId++)           // 2 righe per poll, fino alla riga 6
                batch.add(new SourceRecord(PARTIZIONE, Map.of("prossimo", prossimoId + 1),
                    "cdc.orders", null, "riga-" + prossimoId));
            return batch;
        }
        @Override public void stop() {}
    }

    static final Map<Map<String, String>, Map<String, Object>> store = new HashMap<>();   // offset persistiti
    static final List<Object> topic = new ArrayList<>();                                   // cosa arriva su Kafka

    static TabellaTask nuovoTask() {
        TabellaTask t = new TabellaTask();
        t.initialize(new SourceTaskContext() {
            @Override public Map<String, String> configs() { return Map.of(); }
            @Override public org.apache.kafka.common.metrics.PluginMetrics pluginMetrics() { return null; }   // non usato qui
            @Override public OffsetStorageReader offsetStorageReader() {
                return new OffsetStorageReader() {
                    @Override public <T> Map<String, Object> offset(Map<String, T> p) { return store.get(p); }
                    @Override public <T> Map<Map<String, T>, Map<String, Object>> offsets(Collection<Map<String, T>> ps) {
                        return Map.of();
                    }
                };
            }
        });
        t.start(Map.of());
        return t;
    }

    @SuppressWarnings("unchecked")
    static void scrivi(List<SourceRecord> batch, boolean salvaOffset) {
        for (SourceRecord r : batch) topic.add(r.value());                   // 1) il record arriva su Kafka
        if (salvaOffset && !batch.isEmpty()) {                               // 2) solo poi l'offset viene persistito
            SourceRecord ultimo = batch.get(batch.size() - 1);
            store.put((Map<String, String>) ultimo.sourcePartition(), (Map<String, Object>) ultimo.sourceOffset());
        }
    }

    public static void main(String[] args) throws Exception {
        TabellaTask t1 = nuovoTask();
        scrivi(t1.poll(), true);           // righe 1-2: scritte e offset salvato
        scrivi(t1.poll(), false);          // righe 3-4: scritte, ma il worker cade prima del flush degli offset
        System.out.println("prima del crash: " + topic + ", offset salvato: " + store);

        TabellaTask t2 = nuovoTask();      // riavvio: legge l'offset persistito
        List<SourceRecord> b;
        while (!(b = t2.poll()).isEmpty()) scrivi(b, true);
        System.out.println("dopo il riavvio: " + topic);
    }
}

// Output:
// prima del crash: [riga-1, riga-2, riga-3, riga-4], offset salvato: {{tabella=orders}={prossimo=3}}
// dopo il riavvio: [riga-1, riga-2, riga-3, riga-4, riga-3, riga-4, riga-5, riga-6]

Il task legge due righe per volta e dichiara come offset «prossimo = id + 1» (stessa regola del +1 degli offset del consumatore, vista nell'articolo sui fondamenti). Il primo lotto (righe 1 e 2) viene scritto e il suo offset salvato; il secondo (3 e 4) viene scritto ma il worker cade prima del salvataggio, che nel worker vero avviene a intervalli (offset.flush.interval.ms, 60000 ms di default). Alla ripartenza il task legge l'offset persistito, cioè 3, e rilegge le righe 3 e 4: il topic finisce con riga-3 e riga-4 due volte. È il comportamento at-least-once di un source connector. Connect offre exactly-once per i source connector dalla 3.3, ma la documentazione avverte che dipende dal tipo di connector, che richiede la modalità distribuita e che si abilita con exactly.once.source.support (default disabled); non lo dare per scontato, e verifica nella documentazione del singolo connector se lo supporta.

Debezium per PostgreSQL: leggere il log delle transazioni

PostgreSQL scrive ogni modifica nel write-ahead log (WAL) prima di applicarla. Con wal_level=logical il WAL contiene abbastanza informazioni da ricostruire le modifiche riga per riga, e Debezium si collega come client di replicazione tramite uno slot di replica e il plugin pgoutput, che è incluso in PostgreSQL. Ogni modifica diventa un evento con i campi before, after e op, dove op vale c (insert), u (update), d (delete) o r (lettura, per le righe dello snapshot iniziale, come mostra la documentazione).

Qui sotto c'è la preparazione del database. Due scelte meritano attenzione. La publication è creata a mano e limitata alle tabelle che servono: se lasci il valore predefinito di publication.autocreate.mode, che la documentazione dice essere all_tables, il connector prova a crearne una per tutte le tabelle e fallisce senza i permessi. E REPLICA IDENTITY FULL è facoltativa ma cambia ciò che vedi: con l'identità predefinita, per UPDATE e DELETE il campo before contiene al massimo le colonne della chiave, perché la documentazione fa dipendere il contenuto dei valori precedenti da questa impostazione.

-- Schema da adattare (non eseguito: serve un PostgreSQL con permessi di amministrazione).
-- 1. postgresql.conf (richiede un riavvio)
--    wal_level = logical
--    max_wal_senders = 10
--    max_replication_slots = 10

-- 2. Un ruolo dedicato, con i soli permessi di replica e lettura
CREATE ROLE debezium_user WITH LOGIN REPLICATION PASSWORD 'cambiami';
GRANT SELECT ON public.orders TO debezium_user;

-- 3. La publication, creata a mano e limitata alle tabelle che servono
CREATE PUBLICATION debezium_pub FOR TABLE public.orders;

-- 4. Facoltativo: far arrivare nel campo "before" tutte le colonne delle UPDATE e DELETE
ALTER TABLE public.orders REPLICA IDENTITY FULL;

Il ruolo ha solo i permessi di replica e di lettura sulle tabelle catturate, non quelli di un amministratore: se le credenziali del connector fuggono, il danno è limitato. Il costo di REPLICA IDENTITY FULL è che ogni UPDATE e DELETE scrive più dati nel WAL, quindi attivala solo sulle tabelle in cui ti serve il valore precedente. Il file di configurazione del connector, che si invia alla REST API con POST /connectors, è il seguente.

{
  "name": "postgres-orders-source",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",
    "database.hostname": "postgres-host",
    "database.port": "5432",
    "database.user": "debezium_user",
    "database.password": "${dirProvider:/etc/kafka-connect/secrets:db.password}",
    "database.dbname": "ecommerce",
    "topic.prefix": "cdc",

    "plugin.name": "pgoutput",
    "publication.name": "debezium_pub",
    "publication.autocreate.mode": "disabled",
    "slot.name": "debezium_slot",
    "table.include.list": "public.orders",
    "heartbeat.interval.ms": "10000",

    "transforms": "unwrap",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.delete.tombstone.handling.mode": "rewrite",

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

Le proprietà che contano sono cinque. topic.prefix è il prefisso dei topic (qui cdc.public.orders) e sostituisce il vecchio database.server.name, che la documentazione attuale non cita più. plugin.name va impostato a pgoutput perché il valore predefinito documentato è decoderbufs, un plugin da installare a parte. slot.name e publication.name identificano lo slot e la publication: lo slot è dove PostgreSQL ricorda la posizione del connector, quindi cambiarne il nome significa perderla. heartbeat.interval.ms serve quando le tabelle catturate cambiano di rado ma il database è attivo: senza heartbeat lo slot non avanza e il WAL cresce (lo dice la sezione sul consumo di disco della documentazione). delete.tombstone.handling.mode=rewrite dell'SMT ExtractNewRecordState appiattisce l'evento nel solo stato «dopo» e, per le cancellazioni, produce il record con il campo __deleted: "true" invece di un tombstone. Le password non vanno nel file: il segnaposto ${dirProvider:...} usa il meccanismo dei config provider di Kafka, che va dichiarato nella configurazione del worker (config.providers=dirProvider e la sua classe, come nella documentazione «Configuration Providers»).

Le trasformazioni leggere (SMT), eseguite davvero

Una Single Message Transform modifica ogni record mentre attraversa Connect, senza un'applicazione Streams a valle. Si dichiarano in ordine con transforms e si applicano in quell'ordine. Il listato configura tre trasformazioni vere di Kafka Connect, con le stesse proprietà che metteresti nel JSON, e le applica a un record:

import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.InsertField;
import org.apache.kafka.connect.transforms.RegexRouter;
import org.apache.kafka.connect.transforms.ReplaceField;
import java.util.*;

public class CatenaSmt {
    public static void main(String[] args) {
        // Le tre trasformazioni sono le classi vere di Kafka Connect, configurate come nel file del connector.
        Transformation<SourceRecord> rinomina = new ReplaceField.Value<>();
        rinomina.configure(Map.of("renames", "user_id:userId,order_id:orderId"));
        Transformation<SourceRecord> aggiungiOra = new InsertField.Value<>();
        aggiungiOra.configure(Map.of("timestamp.field", "ingestion_ts"));
        Transformation<SourceRecord> instrada = new RegexRouter<>();
        instrada.configure(Map.of("regex", "cdc\\.public\\.(.*)", "replacement", "db.changes.$1"));

        Map<String, Object> valore = new LinkedHashMap<>();
        valore.put("order_id", 123);
        valore.put("user_id", "user-42");
        valore.put("amount", 99.99);
        SourceRecord r = new SourceRecord(Map.of("db", "ecommerce"), Map.of("lsn", 1000L),
            "cdc.public.orders", null, null, null, null, valore, 1_700_000_000_000L);

        System.out.println("prima:   " + r.topic() + " " + r.value());
        SourceRecord dopo = instrada.apply(aggiungiOra.apply(rinomina.apply(r)));
        System.out.println("dopo:    " + dopo.topic() + " " + dopo.value());
    }
}

// Output:
// prima:   cdc.public.orders {order_id=123, user_id=user-42, amount=99.99}
// dopo:    db.changes.orders {amount=99.99, orderId=123, userId=user-42, ingestion_ts=1700000000000}

Il campo order_id è diventato orderId e user_id è diventato userId (ReplaceField), il topic cdc.public.orders è diventato db.changes.orders (RegexRouter) e il campo ingestion_ts è comparso (InsertField). Attenzione al nome di quest'ultimo: timestamp.field inserisce il timestamp del record (nel listato 1700000000000, quello che ho dato al SourceRecord), non l'ora in cui il record entra in Kafka. Le chiamate sono annidate dall'interno verso l'esterno nell'ordine rinomina, aggiungiOra, instrada, lo stesso della lista transforms: invertirle cambia il risultato quando una trasformazione dipende dall'altra (se il router agisse per primo, un'altra trasformazione che cerca il vecchio nome del topic non lo troverebbe).

Per applicare una trasformazione solo ad alcuni record si usano i predicati, definiti a parte e agganciati con la proprietà implicita predicate (e negate per invertirli). La combinazione tipica è Filter più un predicato, come nella documentazione:

# Schema da adattare (dalla documentazione di Connect): scartare i record di un topic con Filter + predicato
transforms = Scarta
transforms.Scarta.type = org.apache.kafka.connect.transforms.Filter
transforms.Scarta.predicate = EDiTest
predicates = EDiTest
predicates.EDiTest.type = org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.EDiTest.pattern = .*\.test

Qui Filter scarta il record quando il predicato è vero, e il predicato TopicNameMatches ha come unica proprietà pattern. I predicati inclusi sono tre: TopicNameMatches, HasHeaderKey e RecordIsTombstone. Non usare Filter da solo né con una proprietà condition: senza predicato scarta ogni record.

Errori e disco: i due modi in cui un connector fa male

Per default un connector è «fail fast»: errors.tolerance=none e errors.retry.timeout=0, cioè nessun tentativo e arresto al primo errore. Puoi allargare la tolleranza (errors.tolerance=all), far ritentare per un tempo (errors.retry.timeout) e registrare gli errori nel log. La dead letter queue di Connect (errors.deadletterqueue.topic.name) vale però per i sink connector: la documentazione la descrive per gli errori in conversione, trasformazione o nel sink; non usarla come rete di sicurezza di un source Debezium. Per scegliere che cosa fare dei messaggi che non riesci a elaborare, vedi l'articolo sulle dead letter queue.

Il rischio più serio di Debezium per PostgreSQL non sta in Kafka ma nel database. Lo slot di replica trattiene il WAL finché il connector non l'ha consumato: la documentazione di Debezium avverte che gli slot vanno monitorati da vicino per evitare un consumo eccessivo di disco. Se il connector resta fermo per giorni, il WAL si accumula. PostgreSQL ha il parametro max_slot_wal_keep_size, il cui default è -1, cioè nessun limite: con un valore positivo lo slot, se resta troppo indietro, passa a unreserved e poi a lost, e non è più utilizzabile (documentazione di pg_replication_slots). Scegliere il limite significa scegliere cosa perdere: il disco del database oppure la continuità del flusso, perché dopo uno slot perso il connector dovrà rifare lo snapshot.

-- Schema da adattare (non eseguito): stato dello slot di replica e WAL che trattiene
SELECT slot_name,
       active,
       wal_status,                                   -- reserved | extended | unreserved | lost
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS wal_trattenuto
FROM pg_replication_slots
WHERE slot_name = 'debezium_slot';

La query mostra per lo slot se è attivo, il suo wal_status e quanto WAL trattiene dalla posizione restart_lsn. Va messa dietro un allarme: quando wal_status smette di essere reserved o la dimensione cresce a ogni controllo, il connector non tiene il passo o è fermo. Non indico una soglia in byte: dipende dalla dimensione del tuo disco e dal ritmo di scrittura, e la soglia giusta è una frazione dello spazio libero.

Che cosa sai fare adesso

Sai spiegare che cosa fanno worker, connector e task, e dove Connect ricorda la posizione. Sai perché un source connector è at-least-once e quando il duplicato compare. Sai preparare PostgreSQL per Debezium e perché la publication e REPLICA IDENTITY contano, scrivere un connector con topic.prefix e pgoutput, comporre una catena di SMT e filtrare con un predicato. Soprattutto sai che lo slot di replica è un debito di disco: va monitorato e limitato. Per decidere il formato dei messaggi che il connector produce, passa allo Schema Registry; per vedere come questi eventi diventano analisi, all'articolo su Flink. Il pattern che usa gli stessi eventi per pubblicare in modo atomico è descritto nell'articolo sull'outbox. Se arrivi da un altro punto della serie, i collegamenti sono: Kafka Streams.