07 - Kafka Connect: Source/Sink Connector, Debezium CDC and DB Integration

Every change to a table must reach Kafka, without writing a producer

The orders database changes all the time: new rows, updated statuses, deletions. Other services need to know what changed, but querying the table at intervals (polling) has two known flaws: it costs pointless reads, and it never sees a deletion, because a deleted row can no longer be read. Writing by hand the producer that copies the changes means solving on your own problems that are already solved: restarting without losing your position, splitting the work, handling errors.

Kafka Connect is the runtime that solves this part: you configure a connector with a JSON file, it runs it, remembers where it got to and spreads it across several processes. To read the changes of PostgreSQL there is Debezium, which does not query the tables but reads the database's transaction log. In this article you see three things with executed code, using the real Kafka Connect 4.4.0 classes (connect-api, connect-transforms, Maven Central, SHA-1 verified): how a task remembers its own position, how a chain of transformations works and what happens after a crash. Debezium and PostgreSQL were not run in this environment: the configuration files and the SQL are templates to adapt, checked against the Debezium documentation ("stable" version) and the Kafka 4.4 documentation, both consulted on 2026-11-12.

Worker, connector and task: who does what

A worker is a JVM process that runs connectors. The connector is the definition of the integration ("copy the orders table to Kafka") and decides how many tasks to split it into; the task is the unit that does the actual work. The maximum number is given by tasks.max (default 1), with the documentation's caveat that the connector may create fewer tasks if it cannot reach that parallelism: the Debezium connector for PostgreSQL, its documentation says, "always uses a single task" and ignores that value, because the transaction log is a single stream. The format of the bytes written to Kafka is decided not by the connector but by the converters (key.converter, value.converter), so that any connector can work with any format.

The two execution modes, according to the user guide, are standalone (a single process, suitable for example for collecting log files, but with no fault tolerance) and distributed, where several workers form a cluster that spreads the work automatically. In distributed mode, configurations, offsets and task statuses live in three Kafka topics (config.storage.topic, offset.storage.topic, status.storage.topic), which the documentation recommends creating by hand. Connectors are managed through the worker's REST API: list with GET /connectors, create with POST /connectors, status with GET /connectors/{name}/status, plus pause, resume and restart.

Offsets: how a connector remembers where it got to

Every record that a source connector produces carries two small maps: a source partition (which source it is, for example the table) and a source offset (how far it has been read). Connect periodically saves the last offset in the offsets topic and hands it back to the task when it restarts. The listing uses the real SourceTask and SourceRecord classes of the Connect API and only simulates the worker's job, which here is a small in-memory loop: it reads a batch, writes it to the "topic", and only afterwards saves the offset.

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

public class OffsetSorgente {
    // A textbook source task: it reads numbered rows and tells Connect how far it got.
    static class TabellaTask extends SourceTask {
        static final Map<String, String> PARTIZIONE = Map.of("table", "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("next");      // resumes from where it left off
        }
        @Override public List<SourceRecord> poll() {
            List<SourceRecord> batch = new ArrayList<>();
            for (int i = 0; i < 2 && prossimoId <= 6; i++, prossimoId++)           // 2 rows per poll, up to row 6
                batch.add(new SourceRecord(PARTIZIONE, Map.of("next", prossimoId + 1),
                    "cdc.orders", null, "row-" + prossimoId));
            return batch;
        }
        @Override public void stop() {}
    }

    static final Map<Map<String, String>, Map<String, Object>> store = new HashMap<>();   // persisted offsets
    static final List<Object> topic = new ArrayList<>();                                   // what reaches 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; }   // not used here
            @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) the record reaches Kafka
        if (salvaOffset && !batch.isEmpty()) {                               // 2) only then is the offset persisted
            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);           // rows 1-2: written and offset saved
        scrivi(t1.poll(), false);          // rows 3-4: written, but the worker crashes before the offsets flush
        System.out.println("before the crash: " + topic + ", saved offset: " + store);

        TabellaTask t2 = nuovoTask();      // restart: reads the persisted offset
        List<SourceRecord> b;
        while (!(b = t2.poll()).isEmpty()) scrivi(b, true);
        System.out.println("after the restart: " + topic);
    }
}

// Output:
// before the crash: [row-1, row-2, row-3, row-4], saved offset: {{table=orders}={next=3}}
// after the restart: [row-1, row-2, row-3, row-4, row-3, row-4, row-5, row-6]

The task reads two rows at a time and declares as its offset "next = id + 1" (the same +1 rule as consumer offsets, seen in the article on the fundamentals). The first batch (rows 1 and 2) is written and its offset saved; the second (3 and 4) is written but the worker crashes before the save, which in the real worker happens at intervals (offset.flush.interval.ms, 60000 ms by default). On restart the task reads the persisted offset, that is 3, and rereads rows 3 and 4: the topic ends up with row-3 and row-4 twice. This is the at-least-once behavior of a source connector. Connect offers exactly-once for source connectors since 3.3, but the documentation warns that it depends on the type of connector, that it requires distributed mode and that it is enabled with exactly.once.source.support (default disabled); do not take it for granted, and check in the documentation of the individual connector whether it supports it.

Debezium for PostgreSQL: reading the transaction log

PostgreSQL writes every change to the write-ahead log (WAL) before applying it. With wal_level=logical the WAL contains enough information to rebuild the changes row by row, and Debezium connects as a replication client through a replication slot and the pgoutput plugin, which is included in PostgreSQL. Every change becomes an event with the fields before, after and op, where op is c (insert), u (update), d (delete) or r (read, for the rows of the initial snapshot, as the documentation shows).

Below is the database preparation. Two choices deserve attention. The publication is created by hand and limited to the tables you need: if you leave the default value of publication.autocreate.mode, which the documentation says is all_tables, the connector tries to create one for all the tables and fails without the permissions. And REPLICA IDENTITY FULL is optional but changes what you see: with the default identity, for UPDATE and DELETE the before field contains at most the key columns, because the documentation makes the content of the previous values depend on this setting.

-- Template to adapt (not run: it needs a PostgreSQL with administrator permissions).
-- 1. postgresql.conf (requires a restart)
--    wal_level = logical
--    max_wal_senders = 10
--    max_replication_slots = 10

-- 2. A dedicated role, with replication and read permissions only
CREATE ROLE debezium_user WITH LOGIN REPLICATION PASSWORD 'changeme';
GRANT SELECT ON public.orders TO debezium_user;

-- 3. The publication, created by hand and limited to the tables you need
CREATE PUBLICATION debezium_pub FOR TABLE public.orders;

-- 4. Optional: make the "before" field carry all the columns of UPDATE and DELETE
ALTER TABLE public.orders REPLICA IDENTITY FULL;

The role has only replication and read permissions on the captured tables, not those of an administrator: if the connector's credentials leak, the damage is limited. The cost of REPLICA IDENTITY FULL is that every UPDATE and DELETE writes more data to the WAL, so enable it only on the tables where you need the previous value. The connector configuration file, which is sent to the REST API with POST /connectors, is the following.

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

Five properties matter. topic.prefix is the prefix of the topics (here cdc.public.orders) and replaces the old database.server.name, which the current documentation no longer mentions. plugin.name must be set to pgoutput because the documented default is decoderbufs, a plugin you have to install separately. slot.name and publication.name identify the slot and the publication: the slot is where PostgreSQL remembers the connector's position, so changing its name means losing it. heartbeat.interval.ms is needed when the captured tables change rarely but the database is active: without a heartbeat the slot does not advance and the WAL grows (the documentation's section on disk consumption says so). delete.tombstone.handling.mode=rewrite of the ExtractNewRecordState SMT flattens the event into the "after" state alone and, for deletions, produces the record with the field __deleted: "true" instead of a tombstone. Passwords do not belong in the file: the ${dirProvider:...} placeholder uses Kafka's config provider mechanism, which must be declared in the worker configuration (config.providers=dirProvider and its class, as in the "Configuration Providers" documentation).

Lightweight transformations (SMTs), really executed

A Single Message Transform modifies each record as it passes through Connect, without a Streams application downstream. They are declared in order with transforms and applied in that order. The listing configures three real Kafka Connect transformations, with the same properties you would put in the JSON, and applies them to a 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) {
        // The three transformations are the real Kafka Connect classes, configured as in the connector file.
        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("before:  " + r.topic() + " " + r.value());
        SourceRecord dopo = instrada.apply(aggiungiOra.apply(rinomina.apply(r)));
        System.out.println("after:   " + dopo.topic() + " " + dopo.value());
    }
}

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

The field order_id became orderId and user_id became userId (ReplaceField), the topic cdc.public.orders became db.changes.orders (RegexRouter) and the field ingestion_ts appeared (InsertField). Mind the name of the latter: timestamp.field inserts the record's timestamp (in the listing 1700000000000, the one I gave to the SourceRecord), not the time at which the record enters Kafka. The calls are nested from the inside out in the order rinomina, aggiungiOra, instrada (Italian identifiers for rename, add time, route), the same as the transforms list: reversing them changes the result when one transformation depends on another (if the router acted first, another transformation looking for the old topic name would not find it).

To apply a transformation to only some records you use predicates, defined separately and attached with the implicit predicate property (and negate to invert them). The typical combination is Filter plus a predicate, as in the documentation:

# Template to adapt (from the Connect documentation): drop the records of a topic with Filter + predicate
transforms = Drop
transforms.Drop.type = org.apache.kafka.connect.transforms.Filter
transforms.Drop.predicate = IsTest
predicates = IsTest
predicates.IsTest.type = org.apache.kafka.connect.transforms.predicates.TopicNameMatches
predicates.IsTest.pattern = .*\.test

Here Filter drops the record when the predicate is true, and the TopicNameMatches predicate has pattern as its only property. Three predicates are included: TopicNameMatches, HasHeaderKey and RecordIsTombstone. Do not use Filter on its own or with a condition property: without a predicate it drops every record.

Errors and disk: the two ways a connector hurts

By default a connector is "fail fast": errors.tolerance=none and errors.retry.timeout=0, that is no retries and a stop at the first error. You can widen the tolerance (errors.tolerance=all), have it retry for a while (errors.retry.timeout) and log the errors. The Connect dead letter queue (errors.deadletterqueue.topic.name), however, applies to sink connectors: the documentation describes it for errors in conversion, transformation or in the sink; do not use it as the safety net of a Debezium source. To choose what to do with the messages you cannot process, see the article on dead letter queues.

The most serious risk of Debezium for PostgreSQL is not in Kafka but in the database. The replication slot holds back the WAL until the connector has consumed it: the Debezium documentation warns that slots must be monitored closely to avoid excessive disk consumption. If the connector stays down for days, the WAL piles up. PostgreSQL has the parameter max_slot_wal_keep_size, whose default is -1, that is no limit: with a positive value the slot, if it falls too far behind, goes to unreserved and then to lost, and is no longer usable (documentation of pg_replication_slots). Choosing the limit means choosing what to lose: the database disk or the continuity of the stream, because after a lost slot the connector will have to redo the snapshot.

-- Template to adapt (not run): state of the replication slot and the WAL it retains
SELECT slot_name,
       active,
       wal_status,                                   -- reserved | extended | unreserved | lost
       pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS retained_wal
FROM pg_replication_slots
WHERE slot_name = 'debezium_slot';

The query shows for the slot whether it is active, its wal_status and how much WAL it retains from the restart_lsn position. Put an alarm behind it: when wal_status stops being reserved or the size grows at every check, the connector is not keeping up or is stopped. I give no threshold in bytes: it depends on the size of your disk and the write rate, and the right threshold is a fraction of the free space.

What you can do now

You can explain what workers, connectors and tasks do, and where Connect remembers the position. You know why a source connector is at-least-once and when the duplicate appears. You can prepare PostgreSQL for Debezium and say why the publication and REPLICA IDENTITY matter, write a connector with topic.prefix and pgoutput, compose a chain of SMTs and filter with a predicate. Above all you know that the replication slot is a disk debt: it must be monitored and limited. To decide the format of the messages the connector produces, move on to the Schema Registry; to see how these events become analytics, go to the article on Flink. The pattern that uses the same events to publish atomically is described in the article on the outbox. If you come from another point of the series, the links are: Kafka Streams.