Kafka Connect Deep Dive

May 24, 2026 35 min read Data & Databases

kafkakafka-connectdebeziumcdcstrimzikubernetesstreamingdata-engineeringschema-registrydistributed-systems

Contents

Kafka Connect is the data integration layer of the Kafka ecosystem. It bridges the gap between Kafka and every other system you run — databases, object stores, search engines, data warehouses, SaaS APIs — using a framework of pluggable, scalable, fault-tolerant connectors. It looks deceptively simple in the docs: deploy a JAR, POST a JSON config, data flows. Then you hit production, and the questions start. Why did the connector pause? What is a replication slot and why is my Postgres disk filling up? How do I guarantee exactly-once delivery for a source connector? Why did schema evolution break my consumers at 2 AM?

This guide covers all of it — architecture, CDC with Debezium, transforms, exactly-once semantics, Schema Registry, error handling, Kubernetes deployment with Strimzi, and monitoring. All configuration examples are tested against current versions: Debezium 3.5.0.Final (released March 2026), Strimzi 0.51, and Apache Kafka 4.1.

Architecture

What Kafka Connect Is

Kafka Connect is a JVM process (a worker) that runs connectors. Each connector is a plugin — a JAR on the worker’s classpath — that implements the Connect API. The framework handles everything else: distributed coordination, offset storage, task lifecycle, REST management, error handling, and schema conversion. You write (or download) the connector logic; Connect handles the operations.

There are two flavors of connector:

Source connectors pull data from an external system and write records to Kafka topics.
Sink connectors consume records from Kafka topics and write them to an external system.

Workers: Standalone vs Distributed

Standalone mode runs a single worker process. Offsets and configs are stored in local files. It exists for development and one-off migrations, not production. Never use it for anything that needs to survive a process restart reliably.

Distributed mode is what you run everywhere else. Workers form a cluster via a Kafka consumer group. The cluster state — connector configurations, task offsets, connector/task statuses — is stored in three compacted Kafka topics. Any worker in the group can serve the REST API, and requests are forwarded to the right worker transparently.

┌──────────────────────────────────────────────────────────────┐
│                   Kafka Connect Cluster                       │
│                                                              │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐             │
│  │  Worker 1  │  │  Worker 2  │  │  Worker 3  │             │
│  │            │  │            │  │            │             │
│  │ Task A/0   │  │ Task A/1   │  │ Task A/2   │             │
│  │ Task B/0   │  │            │  │ Task B/1   │             │
│  │            │  │            │  │            │             │
│  │ REST :8083 │  │ REST :8083 │  │ REST :8083 │             │
│  └────────────┘  └────────────┘  └────────────┘             │
│         │               │               │                    │
│         └───────────────┴───────────────┘                    │
│                         │                                    │
│              ┌──────────▼──────────┐                         │
│              │    Kafka Broker     │                         │
│              │                    │                         │
│              │  connect-configs   │  ← connector configs    │
│              │  connect-offsets   │  ← source offsets       │
│              │  connect-statuses  │  ← connector/task state │
│              └────────────────────┘                         │
└──────────────────────────────────────────────────────────────┘

Internal Topics

Three internal topics are created automatically on first startup. Their names are configured via:

1
2
3


config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statuses

Recommended settings for production:

Topic	Partitions	Replication Factor	Cleanup Policy
`connect-configs`	1	3	compact
`connect-offsets`	25	3	compact
`connect-statuses`	5	3	compact,delete

The config topic has exactly one partition because it is ordered (serialized writes). The offset topic benefits from more partitions for throughput when running many connectors. Create these topics manually before starting workers if you need specific configurations — Connect will create them with its defaults if they don’t exist, and those defaults (replication factor 1) are wrong for production.

Connectors, Tasks, and Parallelism

A connector is a logical entity with a name and a configuration. It is responsible for determining how many tasks to create and what each task does. A task is the actual unit of work — a thread (or lightweight concurrency unit) doing the I/O.

tasks.max sets the ceiling on parallelism. The connector determines the actual task count based on what makes sense for the source. A Debezium connector for a single database will always use one task regardless of tasks.max, because CDC from a single transaction log is inherently sequential. A JDBC source connector polling 20 tables with tasks.max=5 will create 5 tasks covering 4 tables each.

Rebalancing

Workers use Kafka’s consumer group protocol to elect a leader and distribute work. When a worker joins, leaves, or fails, a rebalance happens. Since Kafka 2.3, Connect uses Incremental Cooperative Rebalancing (KIP-415), which avoids the stop-the-world full rebalance of the old eager protocol. In cooperative rebalancing, workers give up only the tasks that need to move rather than stopping everything.

During rebalance, the leader worker computes a new assignment and communicates it via the consumer group protocol. The REST API returns HTTP 409 (Conflict) on connector operations during an active rebalance.

REST API Forwarding

Every worker exposes a REST API on port 8083. You can POST to any worker and it will forward the request to the correct worker if needed. The rest.advertised.host.name and rest.advertised.port settings control how workers advertise themselves to each other. In Kubernetes, set rest.advertised.host.name to the pod’s IP (use the downward API) so internal forwarding uses pod IPs rather than hostnames that might not resolve correctly.

Worker Configuration

A minimal distributed worker connect-distributed.properties:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39


# Kafka broker(s)
bootstrap.servers=kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092

# Cluster identity — ALL workers in the same cluster must share this
group.id=connect-cluster-prod

# Internal storage topics
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-statuses

config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3

# Serialization for the worker-level key/value
# Each connector can override these
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

# Plugin discovery — directories (or JARs) containing connector plugins
plugin.path=/usr/share/java,/usr/share/confluent-hub-components,/opt/connectors

# REST API
listeners=HTTP://0.0.0.0:8083
rest.advertised.host.name=worker-1.connect.svc.cluster.local
rest.advertised.port=8083

# Exactly-once source support (Kafka 3.3+)
exactly.once.source.support=enabled

# Producer settings for source connectors (EOS requires idempotent producer)
producer.enable.idempotence=true
producer.acks=all

# Consumer settings for sink connectors
consumer.isolation.level=read_committed

The REST API

Kafka Connect’s REST API is the primary operational interface. All operations are idempotent except connector creation (POST).

Creating a Connector

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16


curl -s -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "my-connector",
    "config": {
      "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
      "tasks.max": "1",
      "database.hostname": "postgres",
      "database.port": "5432",
      "database.user": "debezium",
      "database.password": "debezium",
      "database.dbname": "mydb",
      "topic.prefix": "pg",
      "plugin.name": "pgoutput"
    }
  }' | jq .

Checking Status

1
2
3
4
5
6
7
8
9


# Connector status
curl -s http://connect:8083/connectors/my-connector/status | jq .

# All connectors with status (Kafka 2.3+)
curl -s 'http://connect:8083/connectors?expand=status' | jq .

# Individual task status
curl -s http://connect:8083/connectors/my-connector/status | \
  jq '.tasks[] | select(.state != "RUNNING")'

Response structure:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15


{
  "name": "my-connector",
  "connector": {
    "state": "RUNNING",
    "worker_id": "10.0.0.1:8083"
  },
  "tasks": [
    {
      "id": 0,
      "state": "RUNNING",
      "worker_id": "10.0.0.1:8083"
    }
  ],
  "type": "source"
}

Updating Configuration

1
2
3
4


# Full config replace (PUT is idempotent)
curl -s -X PUT http://connect:8083/connectors/my-connector/config \
  -H "Content-Type: application/json" \
  -d '{ "connector.class": "...", "tasks.max": "2", ... }' | jq .

Lifecycle Operations

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22


# Pause (stops task I/O but keeps the connector registered)
curl -s -X PUT http://connect:8083/connectors/my-connector/pause

# Resume
curl -s -X PUT http://connect:8083/connectors/my-connector/resume

# Restart the connector process (not the tasks)
curl -s -X POST http://connect:8083/connectors/my-connector/restart

# Restart a specific task
curl -s -X POST http://connect:8083/connectors/my-connector/tasks/0/restart

# Delete
curl -s -X DELETE http://connect:8083/connectors/my-connector

# List available connector plugins
curl -s http://connect:8083/connector-plugins | jq '.[].class'

# Validate a config before deploying
curl -s -X PUT http://connect:8083/connector-plugins/PostgresConnector/config/validate \
  -H "Content-Type: application/json" \
  -d '{ ... config ... }' | jq '.error_count, .configs[].value.errors'

Connector Types and Popular Connectors

Source Connectors

Source connectors read from an external system and write events to Kafka. They implement SourceConnector and SourceTask. The task’s poll() method returns a list of SourceRecord objects each time it is called.

Popular source connectors:

Connector	Class	Use Case
Debezium PostgreSQL	`io.debezium.connector.postgresql.PostgresConnector`	CDC via pgoutput logical replication
Debezium MySQL	`io.debezium.connector.mysql.MySqlConnector`	CDC via binary log
Debezium MongoDB	`io.debezium.connector.mongodb.MongoDbConnector`	CDC via oplog/change streams
Debezium SQL Server	`io.debezium.connector.sqlserver.SqlServerConnector`	CDC via SQL Server CDC feature
JDBC Source	`io.confluent.connect.jdbc.JdbcSourceConnector`	Poll-based ingestion (no CDC)
MirrorMaker2	(built-in)	Cross-cluster Kafka replication

Sink Connectors

Sink connectors consume from Kafka topics and write to external systems. They implement SinkConnector and SinkTask. The task receives batches of SinkRecord objects via put().

Popular sink connectors:

Connector	Class	Use Case
S3 Sink	`io.confluent.connect.s3.S3SinkConnector`	Kafka → S3 data lake
JDBC Sink	`io.confluent.connect.jdbc.JdbcSinkConnector`	Kafka → RDBMS
Elasticsearch/OpenSearch	`io.confluent.connect.elasticsearch.ElasticsearchSinkConnector`	Kafka → search index
HTTP Sink	various	Kafka → REST API
Debezium JDBC Sink	`io.debezium.connector.jdbc.JdbcSinkConnector`	Debezium-native sink, schema-aware
BigQuery	`com.wepay.kafka.connect.bigquery.BigQuerySinkConnector`	Kafka → BigQuery
Snowflake	`com.snowflake.kafka.connector.SnowflakeSinkConnector`	Kafka → Snowflake

Complete Connector Configuration Examples

Debezium PostgreSQL (CDC Source)

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50


{
  "name": "pg-inventory-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "tasks.max": "1",

    "database.hostname": "postgres.db.svc.cluster.local",
    "database.port": "5432",
    "database.user": "debezium",
    "database.password": "${file:/opt/kafka/external-configuration/connector-secrets/pg-password.properties:database.password}",
    "database.dbname": "inventory",

    "topic.prefix": "pg.inventory",
    "plugin.name": "pgoutput",
    "slot.name": "debezium_inventory_slot",
    "publication.name": "dbz_publication",
    "publication.autocreate.mode": "filtered",

    "table.include.list": "public.orders,public.customers,public.products",

    "snapshot.mode": "initial",
    "snapshot.isolation.mode": "repeatable_read",

    "heartbeat.interval.ms": "10000",
    "heartbeat.action.query": "INSERT INTO debezium_heartbeat(ts) VALUES (now()) ON CONFLICT (id) DO UPDATE SET ts = now()",

    "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",
    "transforms.unwrap.add.fields": "op,table,lsn,source.ts_ms",

    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "pg-inventory-cdc-dlq",
    "errors.deadletterqueue.context.headers.enable": "true",

    "max.batch.size": "2048",
    "max.queue.size": "8192",
    "poll.interval.ms": "500",

    "decimal.handling.mode": "double",
    "binary.handling.mode": "base64",
    "time.precision.mode": "connect"
  }
}

JDBC Source Connector

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32


{
  "name": "jdbc-orders-source",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "3",

    "connection.url": "jdbc:postgresql://postgres:5432/mydb",
    "connection.user": "kafka_reader",
    "connection.password": "secret",

    "mode": "timestamp+incrementing",
    "timestamp.column.name": "updated_at",
    "incrementing.column.name": "id",
    "table.whitelist": "orders,order_items,customers",

    "topic.prefix": "jdbc.",
    "poll.interval.ms": "5000",
    "batch.max.rows": "1000",
    "table.poll.interval.ms": "60000",

    "key.converter": "org.apache.kafka.connect.json.JsonConverter",
    "key.converter.schemas.enable": "true",
    "value.converter": "org.apache.kafka.connect.json.JsonConverter",
    "value.converter.schemas.enable": "true",

    "transforms": "createKey,extractInt",
    "transforms.createKey.type": "org.apache.kafka.connect.transforms.ValueToKey",
    "transforms.createKey.fields": "id",
    "transforms.extractInt.type": "org.apache.kafka.connect.transforms.ExtractField$Key",
    "transforms.extractInt.field": "id"
  }
}

S3 Sink Connector

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43


{
  "name": "s3-orders-sink",
  "config": {
    "connector.class": "io.confluent.connect.s3.S3SinkConnector",
    "tasks.max": "4",

    "topics": "pg.inventory.public.orders,pg.inventory.public.customers",
    "topics.dir": "raw",

    "s3.region": "us-east-1",
    "s3.bucket.name": "my-data-lake-bucket",
    "s3.part.size": "5242880",

    "storage.class": "io.confluent.connect.s3.storage.S3Storage",
    "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",
    "parquet.codec": "snappy",

    "flush.size": "10000",
    "rotate.interval.ms": "600000",
    "rotate.schedule.interval.ms": "3600000",

    "partitioner.class": "io.confluent.connect.storage.partitioner.TimeBasedPartitioner",
    "path.format": "'year'=YYYY/'month'=MM/'day'=dd/'hour'=HH",
    "partition.duration.ms": "3600000",
    "locale": "en_US",
    "timezone": "UTC",
    "timestamp.extractor": "RecordField",
    "timestamp.field": "ts_ms",

    "key.converter": "org.apache.kafka.connect.storage.StringConverter",
    "value.converter": "io.confluent.connect.avro.AvroConverter",
    "value.converter.schema.registry.url": "http://schema-registry:8081",

    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "s3-sink-dlq",
    "errors.deadletterqueue.context.headers.enable": "true",
    "errors.log.enable": "true",
    "errors.log.include.messages": "true",

    "s3.ssea.name": "aws:kms",
    "s3.sse.kms.key.id": "arn:aws:kms:us-east-1:123456789:key/my-key-id"
  }
}

Single Message Transforms (SMTs)

SMTs are lightweight, stateless transformations applied to each record as it passes through Connect. They run after the converter deserializes the bytes into a SourceRecord (for source connectors) or before the converter serializes it (for sink connectors).

SMTs are not a stream processing replacement — they cannot join streams, aggregate records, or maintain state. For anything beyond per-record filtering and field manipulation, use Kafka Streams or Flink.

Configuring Transforms

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


{
  "transforms": "rename,addTimestamp,routeByRegion",
  "transforms.rename.type": "org.apache.kafka.connect.transforms.ReplaceField$Value",
  "transforms.rename.renames": "user_id:userId,created_ts:createdAt",
  "transforms.addTimestamp.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addTimestamp.timestamp.field": "ingestTimestamp",
  "transforms.routeByRegion.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.routeByRegion.regex": "events\\.(.+)",
  "transforms.routeByRegion.replacement": "events-us-east.$1"
}

Transforms are applied in the order they are listed in the transforms property. Each transform has a name (you choose it), a type, and any type-specific properties prefixed with the name.

Built-In SMTs Reference

SMT	Class	What It Does
`ExtractField`	`...ExtractField$Key` / `$Value`	Pull a single field out of a struct, making it the new key/value
`ReplaceField`	`...ReplaceField$Value`	Rename, include (whitelist), or exclude (blacklist) fields
`MaskField`	`...MaskField$Value`	Replace field value with type-appropriate zero/null/empty
`Filter`	`...Filter`	Drop records matching a predicate (use with Predicates)
`InsertField`	`...InsertField$Value`	Add a field with a static value, partition, offset, or timestamp
`TimestampConverter`	`...TimestampConverter$Value`	Convert between unix epoch ms, Date, Time, Timestamp, string
`ValueToKey`	`...ValueToKey`	Copy fields from the record value into the key
`RegexRouter`	`...RegexRouter`	Rewrite topic names with a regex replacement
`SetSchemaMetadata`	`...SetSchemaMetadata$Value`	Override schema name or version
`Flatten`	`...Flatten$Value`	Flatten nested structs to top-level with dot-notation field names
`Cast`	`...Cast$Value`	Change field types (STRING → INT32, etc.)
`HoistField`	`...HoistField$Value`	Wrap the entire value in a struct under a named field
`DropHeaders`	`...DropHeaders`	Remove specific headers from the record

For Debezium specifically, the ExtractNewRecordState SMT (bundled with Debezium) is essential — it unwraps the CDC envelope to expose just the after field as the record value.

Using Predicates with Filter

Predicates attach boolean logic to any SMT, making the transform conditional:

1
2
3
4
5
6
7
8
9


{
  "transforms": "dropDeletes",
  "transforms.dropDeletes.type": "org.apache.kafka.connect.transforms.Filter",
  "transforms.dropDeletes.predicate": "isDelete",
  "transforms.dropDeletes.negate": "false",

  "predicates": "isDelete",
  "predicates.isDelete.type": "org.apache.kafka.connect.predicates.RecordIsTombstone"
}

Other built-in predicates: HasHeaderKey, TopicNameMatches.

Writing a Custom SMT

A custom SMT is a JAR that implements org.apache.kafka.connect.transforms.Transformation<R>:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59


package com.example.connect.transforms;

import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.ConnectRecord;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.SchemaBuilder;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.transforms.Transformation;
import org.apache.kafka.connect.transforms.util.SimpleConfig;

import java.util.Map;

public class AddEnvironmentField<R extends ConnectRecord<R>> implements Transformation<R> {

    private static final String ENV_CONFIG = "environment";
    private static final ConfigDef CONFIG_DEF = new ConfigDef()
        .define(ENV_CONFIG, ConfigDef.Type.STRING, "production",
                ConfigDef.Importance.MEDIUM, "Environment label to inject");

    private String environment;

    @Override
    public void configure(Map<String, ?> configs) {
        SimpleConfig config = new SimpleConfig(CONFIG_DEF, configs);
        this.environment = config.getString(ENV_CONFIG);
    }

    @Override
    public R apply(R record) {
        if (record.valueSchema() == null || record.value() == null) {
            return record; // schemaless or tombstone — pass through
        }
        Schema originalSchema = record.valueSchema();
        SchemaBuilder builder = SchemaBuilder.struct().name(originalSchema.name());
        originalSchema.fields().forEach(f -> builder.field(f.name(), f.schema()));
        builder.field("_environment", Schema.STRING_SCHEMA);
        Schema newSchema = builder.build();

        Struct originalValue = (Struct) record.value();
        Struct newValue = new Struct(newSchema);
        originalSchema.fields().forEach(f -> newValue.put(f.name(), originalValue.get(f)));
        newValue.put("_environment", environment);

        return record.newRecord(
            record.topic(), record.kafkaPartition(),
            record.keySchema(), record.key(),
            newSchema, newValue,
            record.timestamp()
        );
    }

    @Override
    public ConfigDef config() {
        return CONFIG_DEF;
    }

    @Override
    public void close() {}
}

Deploy by placing the JAR in your plugin.path directory. Configure it like any other SMT:

1
2
3
4
5


{
  "transforms": "addEnv",
  "transforms.addEnv.type": "com.example.connect.transforms.AddEnvironmentField",
  "transforms.addEnv.environment": "production"
}

Debezium in Depth

Debezium is the most important source connector ecosystem for Kafka. It implements Change Data Capture (CDC) by reading database transaction logs rather than polling tables. This gives you:

Real-time row-level change events (insert, update, delete)
Zero additional load on the database from polling
Exact change ordering per table
A complete history of changes, not just snapshots

As of Debezium 3.5.0.Final (March 2026), supported databases include:

Database	Log Mechanism	Notes
PostgreSQL	Logical replication (pgoutput / decoderbufs)	pgoutput is native since PG 10; no extension needed
MySQL	Binary log (binlog)	Requires `binlog_format=ROW`
MariaDB	Binary log	Separate connector from MySQL since Debezium 2.5
MongoDB	Change streams / oplog	Requires replica set or sharded cluster
SQL Server	SQL Server CDC feature	Requires CDC enabled per-table
Oracle	LogMiner	Requires supplemental logging; significant DBA setup
Db2	ASN Agent / journal	z/OS and LUW
CockroachDB	Changefeeds	Added in 3.5 with snapshot support
Informix	CDC API

The Debezium Envelope Format

Every Debezium event is a Kafka message with a structured envelope value. Understanding this format is essential before deploying Debezium anywhere.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40


{
  "schema": { ... },
  "payload": {
    "before": {
      "id": 1001,
      "first_name": "Sally",
      "last_name": "Thomas",
      "email": "sally.thomas@acme.com"
    },
    "after": {
      "id": 1001,
      "first_name": "Sally",
      "last_name": "Thomas",
      "email": "sally.thomas@example.com"
    },
    "source": {
      "version": "3.5.0.Final",
      "connector": "postgresql",
      "name": "pg.inventory",
      "ts_ms": 1715962800000,
      "ts_us": 1715962800000000,
      "ts_ns": 1715962800000000000,
      "snapshot": "false",
      "db": "inventory",
      "sequence": "[\"24023209\",\"24023456\"]",
      "schema": "public",
      "table": "customers",
      "txId": 555,
      "lsn": 24023456,
      "xmin": null
    },
    "op": "u",
    "ts_ms": 1715962800100,
    "transaction": {
      "id": "555",
      "total_order": 3,
      "data_collection_order": 1
    }
  }
}

op field values:

c — create (INSERT)
u — update (UPDATE)
d — delete (DELETE)
r — read (snapshot row)
t — truncate (table TRUNCATE, PostgreSQL only)
m — message (Debezium heartbeat / logical decoding message)

before is null for inserts. after is null for deletes.

For deletes, Debezium also emits a tombstone record — a message with the same key but a null value — after the delete event. Tombstones signal Kafka’s log compaction to remove all prior messages for that key. You can suppress tombstones with tombstones.on.delete=false, but you should understand why they exist before doing so.

Snapshot Modes

The snapshot is what runs when the connector first starts (or when offset data is absent) to capture the existing state of the database before beginning to stream live changes.

Mode	Behavior
`initial` (default)	Snapshot tables once, then stream changes. Skips snapshot if offsets already exist.
`initial_only`	Snapshot and stop. Does not stream changes. Useful for one-time migrations.
`schema_only`	Capture schema without row data, then stream changes. Use when you only care about events from now on.
`never`	No snapshot. Starts streaming from current log position. Risky — any rows that existed before the connector started will never appear.
`always`	Snapshot every time the connector starts, even if offsets exist. Expensive; rarely correct.
`when_needed`	Snapshot only when the existing offset is no longer valid (e.g. after a slot was dropped).
`no_data`	Capture schema only (alias for `schema_only` in some versions).

For most production deployments, initial is correct. Use schema_only when you’re adding Debezium to a running system where downstream consumers will handle missing history.

PostgreSQL Deep Dive

PostgreSQL CDC uses logical replication, which streams WAL (Write-Ahead Log) events via a replication slot. The slot is a cursor into the WAL that PostgreSQL preserves until Debezium confirms it has consumed and committed up to a given LSN (Log Sequence Number).

Setup requirements:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


-- postgresql.conf
wal_level = logical
max_replication_slots = 5   -- at least 1 per Debezium connector
max_wal_senders = 5

-- Create the Debezium user
CREATE USER debezium REPLICATION LOGIN PASSWORD 'debezium_password';
GRANT SELECT ON ALL TABLES IN SCHEMA public TO debezium;
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO debezium;

-- Create the publication (Debezium can also auto-create this)
CREATE PUBLICATION dbz_publication FOR TABLE public.orders, public.customers;

Offset management:

Debezium stores the current LSN in Kafka Connect’s offset topic. Each record in the offset topic looks like:

1

{"lsn": 24023456, "txId": 555, "ts_usec": 1715962800000000}

The connector advances the confirmed LSN only after the Kafka producer successfully commits the batch of change events. This means the PostgreSQL replication slot holds the WAL from the last confirmed LSN forward.

Replication slot lag — the critical production issue:

If Debezium stops consuming (connector paused, Kafka unavailable, task crashed), the replication slot keeps retaining WAL. PostgreSQL cannot reclaim WAL space past the slot’s restart_lsn. Left unchecked, this fills your disk and causes a PostgreSQL outage.

Monitor slot lag continuously:

1
2
3
4
5
6
7
8


SELECT
    slot_name,
    pg_size_pretty(pg_wal_lsn_diff(pg_current_wal_lsn(), restart_lsn)) AS lag_size,
    active,
    confirmed_flush_lsn,
    restart_lsn
FROM pg_replication_slots
WHERE slot_type = 'logical';

Alert when lag_size exceeds 5 GB. Set max_slot_wal_keep_size = 20GB (PostgreSQL 13+) to automatically invalidate the slot if it falls too far behind — this is a failsafe, not a solution. The real solution is keeping the connector running and monitoring its health.

Heartbeat tables:

On low-traffic databases, the connector’s LSN may not advance because no changes are happening in the monitored tables. The slot still holds WAL from other unrelated activity. Use a heartbeat table to force periodic LSN advancement:

1
2
3
4
5


CREATE TABLE debezium_heartbeat (
    id SERIAL PRIMARY KEY,
    ts TIMESTAMPTZ NOT NULL DEFAULT NOW()
);
GRANT INSERT, UPDATE ON debezium_heartbeat TO debezium;

In the connector config:

1
2


"heartbeat.interval.ms": "10000",
"heartbeat.action.query": "INSERT INTO debezium_heartbeat(ts) VALUES (now()) ON CONFLICT (id) DO UPDATE SET ts = excluded.ts"

On PostgreSQL 14+, you can use pg_logical_emit_message() instead of a heartbeat table to avoid creating extra tables.

MySQL Deep Dive

MySQL CDC uses the binary log (binlog). Configure MySQL:

1
2
3
4
5
6
7


# my.cnf
[mysqld]
server-id         = 1
log_bin           = /var/log/mysql/mysql-bin.log
binlog_format     = ROW
binlog_row_image  = FULL
expire_logs_days  = 7

Create the Debezium user:

1
2
3


CREATE USER 'debezium'@'%' IDENTIFIED BY 'password';
GRANT SELECT, RELOAD, SHOW DATABASES, REPLICATION SLAVE, REPLICATION CLIENT ON *.* TO 'debezium'@'%';
FLUSH PRIVILEGES;

The MySQL connector tracks position via binlog.filename + binlog.position (and optionally GTID if enabled). Unlike PostgreSQL, there is no slot mechanism — MySQL simply doesn’t replay events prior to the connector’s recorded position. If the binlog rotates and the connector hasn’t consumed up to that point, data is lost. Keep your expire_logs_days high enough for your worst-case connector downtime.

Debezium Deployment Modes

Debezium can run in three modes:

Kafka Connect (recommended for production): Deploy as a Kafka Connect source connector. The Connect framework handles distributed operation, offset management, fault tolerance, and scaling. This is the standard mode.
Debezium Server: A standalone process that runs a Debezium engine and can route events to non-Kafka sinks (Amazon Kinesis, Google Pub/Sub, Azure Event Hubs, Redis Streams). Useful when you don’t have Kafka.
Embedded Engine: Run Debezium in-process within your own Java application. Useful for testing and custom integrations. Not recommended for production CDC workloads.

Schema Evolution with Schema Registry

When Debezium (or any connector) produces Avro-serialized events, the message bytes alone are not self-describing. The Confluent Schema Registry stores the schema separately and embeds a 5-byte header (magic byte + 4-byte schema ID) in each message. Consumers look up the schema ID to deserialize.

Kafka message value (bytes):
┌──────┬────────────────┬─────────────────────────────────┐
│ 0x00 │   Schema ID    │         Avro payload             │
│ (1B) │    (4 bytes)   │      (variable length)           │
└──────┴────────────────┴─────────────────────────────────┘

Converter Configuration

1
2
3
4
5
6
7


{
  "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",
  "value.converter.enhanced.avro.schema.support": "true"
}

Other converters available:

io.confluent.connect.json.JsonSchemaConverter — JSON Schema
io.confluent.connect.protobuf.ProtobufConverter — Protobuf
org.apache.kafka.connect.json.JsonConverter — plain JSON with embedded schema (no registry)
org.apache.kafka.connect.storage.StringConverter — raw strings

Compatibility Modes

The Schema Registry enforces compatibility rules when a new schema version is registered. The default is BACKWARD.

Mode	Consumers can use	Producers can use	Allowed changes
`BACKWARD`	New schema	Old schema	Add optional fields (with defaults), remove fields
`FORWARD`	Old schema	New schema	Remove optional fields, add fields
`FULL`	Either	Either	Only add/remove fields with defaults
`NONE`	—	—	Anything (no checking)
`BACKWARD_TRANSITIVE`	New	Any old	Backward-compatible with ALL prior versions
`FORWARD_TRANSITIVE`	Any old	New	Forward-compatible with ALL prior versions
`FULL_TRANSITIVE`	Any	Any	Fully compatible with ALL prior versions

Upgrade order:

BACKWARD → upgrade consumers first, then producers
FORWARD → upgrade producers first, then consumers
FULL → order doesn’t matter

Practical schema evolution rules for Avro + BACKWARD:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23


// V1 schema
{
  "type": "record",
  "name": "Customer",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "email", "type": "string"}
  ]
}

// V2 schema — BACKWARD compatible (added optional field with default)
{
  "type": "record",
  "name": "Customer",
  "fields": [
    {"name": "id", "type": "int"},
    {"name": "email", "type": "string"},
    {"name": "phone", "type": ["null", "string"], "default": null}
  ]
}

// INCOMPATIBLE change — changing a field type
// {"name": "id", "type": "long"}  ← breaks backward compat

Checking Compatibility via Registry API

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14


# Register a schema
curl -s -X POST http://schema-registry:8081/subjects/pg.inventory.public.customers-value/versions \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "{\"type\":\"record\",\"name\":\"Customer\",...}"}'

# Check compatibility before registering
curl -s -X POST http://schema-registry:8081/compatibility/subjects/pg.inventory.public.customers-value/versions/latest \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"schema": "...new schema..."}' | jq .is_compatible

# Set subject-level compatibility
curl -s -X PUT http://schema-registry:8081/config/pg.inventory.public.customers-value \
  -H "Content-Type: application/vnd.schemaregistry.v1+json" \
  -d '{"compatibility": "FULL_TRANSITIVE"}'

Exactly-Once Semantics

The Problem

Without EOS, a source connector can produce duplicate records when a task fails and restarts. The task publishes a batch to Kafka, the Kafka write succeeds, but the task crashes before writing the updated offsets to the offset storage topic. On restart, it replays from the last committed offset — producing duplicates.

For sink connectors, the sink itself must be idempotent for EOS to be meaningful. Kafka can guarantee exactly-once delivery to Kafka topics, but if the downstream system doesn’t support idempotent writes, you still get duplicates at the destination.

EOS for Source Connectors (KIP-618)

Available since Kafka 3.3, exactly-once source support uses Kafka transactions to atomically write source records and their corresponding offsets in a single transaction:

BEGIN TRANSACTION
  → Write records to output topic(s)
  → Write offsets to offset storage topic
COMMIT

If the task fails after the write but before the commit, the transaction is aborted. On restart, the records are never visible to consumers (they read at read_committed isolation). The offset is not advanced. The task replays from the prior committed position.

Worker configuration:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12


# New cluster: set directly to enabled
exactly.once.source.support=enabled

# Rolling upgrade on existing cluster:
# Step 1: set to "preparing" on all workers (rolling restart)
# Step 2: set to "enabled" on all workers (rolling restart)
exactly.once.source.support=preparing   # then: enabled

# Required producer settings
producer.enable.idempotence=true
producer.acks=all
producer.transactional.id.prefix=connect-

EOS source support requires:

Kafka 3.3+ brokers
Distributed mode only (not standalone)
The connector must be able to provide source offsets (nearly all connectors do)
producer.enable.idempotence=true

Connector-level opt-in:

1
2
3


{
  "exactly.once.support": "required"
}

Set to required to fail fast if EOS isn’t available. Set to requested to use EOS when available but degrade gracefully.

EOS for Sink Connectors

Sink EOS is more nuanced. Connect can consume from Kafka with exactly-once guarantees (using isolation.level=read_committed and consumer group offsets committed inside transactions), but the write to the external system must also be idempotent or transactional.

Examples:

JDBC Sink to PostgreSQL: PostgreSQL supports UPSERT (INSERT ... ON CONFLICT DO UPDATE), which makes writes idempotent given a primary key. With this, you get effective end-to-end EOS.
S3 Sink: Object storage writes are naturally idempotent (same key overwrites same file). Partitioned by offset range, the same data written twice results in identical files.
Elasticsearch: Supports idempotent indexing via document ID; duplicates overwrite.

Error Handling and Dead Letter Queues

Error Handling Config

 1
 2
 3
 4
 5
 6
 7
 8
 9
10


{
  "errors.tolerance": "all",
  "errors.retry.delay.max.ms": "60000",
  "errors.retry.timeout": "300000",
  "errors.log.enable": "true",
  "errors.log.include.messages": "true",
  "errors.deadletterqueue.topic.name": "connector-dlq",
  "errors.deadletterqueue.topic.replication.factor": "3",
  "errors.deadletterqueue.context.headers.enable": "true"
}

errors.tolerance:

none (default): Any conversion or transformation error causes the task to fail immediately. The connector enters FAILED state.
all: Errors are silently skipped. If a DLQ is configured, failed records are routed there. Processing continues.

Important limitation: errors.tolerance only applies to conversion and transformation errors — it does not catch connector polling failures (source) or write failures (sink). A JDBC sink that fails to write to the database will still fail the task regardless of errors.tolerance.

Retry settings:

errors.retry.timeout: Total time in ms to retry a failed operation before giving up. -1 means retry forever.
errors.retry.delay.max.ms: Maximum backoff delay between retries (exponential backoff).

DLQ Headers

When errors.deadletterqueue.context.headers.enable=true, each DLQ record gets headers describing the failure:

__connect.errors.topic              = source-topic-name
__connect.errors.partition          = 3
__connect.errors.offset             = 12345
__connect.errors.connector.name     = my-connector
__connect.errors.task.id            = 0
__connect.errors.stage              = VALUE_CONVERTER
__connect.errors.class.name         = org.apache.kafka.connect.errors.DataException
__connect.errors.exception.message  = Failed to deserialize data...
__connect.errors.exception.stacktrace = ...

Distributed Mode at Scale

Capacity Planning

Each worker runs connector tasks as threads inside the JVM. Key limits:

Tasks per worker: There is no hard limit, but each task holds I/O resources (database connections, HTTP connections). 20-50 active tasks per worker is typical for database connectors. A single Debezium task can saturate a worker if the database is high-throughput.
Memory: Allocate at least 4 GB heap for a production worker. Debezium with large batch sizes needs 8+ GB. Each task’s in-memory queue (max.queue.size) also consumes heap.
CPU: Connect workers are mostly I/O-bound, but serialization (especially Avro) adds CPU load.

Internal Topic Best Practices

Create internal topics manually before starting Connect:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17


kafka-topics.sh --bootstrap-server kafka:9092 \
  --create --topic connect-configs \
  --partitions 1 \
  --replication-factor 3 \
  --config cleanup.policy=compact

kafka-topics.sh --bootstrap-server kafka:9092 \
  --create --topic connect-offsets \
  --partitions 25 \
  --replication-factor 3 \
  --config cleanup.policy=compact

kafka-topics.sh --bootstrap-server kafka:9092 \
  --create --topic connect-statuses \
  --partitions 5 \
  --replication-factor 3 \
  --config cleanup.policy=compact,delete

Worker JVM Tuning

1
2
3
4
5
6
7


export KAFKA_HEAP_OPTS="-Xms4g -Xmx8g"
export KAFKA_JVM_PERFORMANCE_OPTS="-server \
  -XX:+UseG1GC \
  -XX:MaxGCPauseMillis=200 \
  -XX:InitiatingHeapOccupancyPercent=35 \
  -XX:+ExplicitGCInvokesConcurrent \
  -Djava.awt.headless=true"

Multiple Plugin Versions (Kafka 4.1)

Kafka 4.1 introduced KIP-891: you can now install multiple versions of the same connector plugin on a single worker. This allows in-place connector upgrades and downgrades without separate Connect clusters.

Plugins are isolated by version using separate classloaders. Specify the plugin version in the connector config:

1
2
3
4


{
  "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
  "plugin.version": "3.5.0.Final"
}

Running on Kubernetes with Strimzi

Strimzi is the CNCF operator for running Kafka (and Kafka Connect) on Kubernetes. Current version: 0.51 (supports Kubernetes 1.30+, includes v1 stable CRD API introduced in 0.49).

KafkaConnect CRD

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71


apiVersion: kafka.strimzi.io/v1
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  namespace: kafka
  annotations:
    # Enable KafkaConnector resource management (recommended)
    strimzi.io/use-connector-resources: "true"
spec:
  version: 4.1.0    # Kafka version
  replicas: 3
  bootstrapServers: my-cluster-kafka-bootstrap:9093

  # Cluster identity
  groupId: my-connect-cluster
  configStorageTopic: connect-cluster-configs
  offsetStorageTopic: connect-cluster-offsets
  statusStorageTopic: connect-cluster-status

  tls:
    trustedCertificates:
      - secretName: my-cluster-cluster-ca-cert
        pattern: "*.crt"

  authentication:
    type: scram-sha-512
    username: connect-user
    passwordSecret:
      secretName: connect-user
      password: password

  config:
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
    config.storage.topic: connect-cluster-configs
    offset.storage.topic: connect-cluster-offsets
    status.storage.topic: connect-cluster-status

    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

    producer.enable.idempotence: "true"
    producer.acks: "all"
    consumer.isolation.level: read_committed
    exactly.once.source.support: enabled

  resources:
    requests:
      memory: 4Gi
      cpu: "1"
    limits:
      memory: 8Gi
      cpu: "4"

  jvmOptions:
    -Xms: 2048m
    -Xmx: 6144m

  # Metrics via JMX exporter
  metricsConfig:
    type: jmxPrometheusExporter
    valueFrom:
      configMapKeyRef:
        name: connect-metrics-config
        key: metrics-config.yml

  # See spec.build section below for automatic image building
  image: my-registry.example.com/my-connect:3.5.0-kafka-4.1

Building Custom Connector Images with spec.build

Instead of manually building Docker images, Strimzi’s spec.build feature has the operator build the image automatically using Kaniko (on Kubernetes) or OpenShift Builds:

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44


apiVersion: kafka.strimzi.io/v1
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  version: 4.1.0
  replicas: 3
  bootstrapServers: my-cluster-kafka-bootstrap:9093
  groupId: my-connect-cluster
  configStorageTopic: connect-cluster-configs
  offsetStorageTopic: connect-cluster-offsets
  statusStorageTopic: connect-cluster-status

  build:
    output:
      type: docker
      image: my-registry.example.com/my-connect:latest
      pushSecret: my-registry-secret   # K8s secret with registry credentials
    plugins:
      - name: debezium-postgres
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-postgres/3.5.0.Final/debezium-connector-postgres-3.5.0.Final-plugin.tar.gz
            sha512sum: "abc123..."   # optional checksum verification
      - name: debezium-mysql
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/3.5.0.Final/debezium-connector-mysql-3.5.0.Final-plugin.tar.gz
      - name: confluent-s3-sink
        artifacts:
          - type: zip
            url: https://d1i4a15mxbxib1.cloudfront.net/api/plugins/confluentinc/kafka-connect-s3/versions/10.5.7/confluentinc-kafka-connect-s3-10.5.7.zip
      - name: confluent-avro-converter
        artifacts:
          - type: maven
            group: io.confluent
            artifact: kafka-connect-avro-converter
            version: "7.7.0"
  config:
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3

When this resource is applied, the Strimzi Cluster Operator:

Creates a build pod with Kaniko
Downloads the specified artifacts
Builds a Docker image with all plugins installed
Pushes to the specified registry
Updates the KafkaConnect deployment to use the new image

No manual Dockerfile required.

KafkaConnector CRD

With strimzi.io/use-connector-resources: "true" on the KafkaConnect resource, connectors are managed via KafkaConnector resources instead of direct REST API calls. The Strimzi operator reconciles KafkaConnector resources to the Connect REST API.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36


apiVersion: kafka.strimzi.io/v1
kind: KafkaConnector
metadata:
  name: pg-inventory-cdc
  namespace: kafka
  labels:
    # REQUIRED: must match the KafkaConnect resource name
    strimzi.io/cluster: my-connect-cluster
spec:
  class: io.debezium.connector.postgresql.PostgresConnector
  tasksMax: 1

  # Set to true to pause the connector (equivalent to PUT /connectors/{name}/pause)
  pause: false

  # autoRestart automatically restarts failed connectors
  autoRestart:
    enabled: true
    maxRestarts: 10

  config:
    database.hostname: postgres.db.svc.cluster.local
    database.port: "5432"
    database.user: debezium
    database.password: "${secrets:kafka/debezium-pg-secret:password}"
    database.dbname: inventory
    topic.prefix: pg.inventory
    plugin.name: pgoutput
    slot.name: debezium_inventory_slot
    publication.name: dbz_publication
    table.include.list: public.orders,public.customers,public.products
    snapshot.mode: initial
    heartbeat.interval.ms: "10000"
    errors.tolerance: all
    errors.deadletterqueue.topic.name: pg-inventory-dlq
    errors.deadletterqueue.context.headers.enable: "true"

Secrets in KafkaConnector configs: Use the ${secrets:namespace/secret-name:key} syntax to reference Kubernetes secrets directly. The Strimzi operator injects the value; the secret never appears in the KafkaConnector YAML or Connect offset storage.

Scaling Connect Workers

Scale the number of workers by changing spec.replicas:

1
2
3


kubectl patch kafkaconnect my-connect-cluster \
  --type=merge \
  -p '{"spec": {"replicas": 5}}'

Strimzi performs a rolling update. Connect’s incremental cooperative rebalancing handles task redistribution with minimal disruption.

Local Development with Docker Compose

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87


version: "3.9"

services:
  kafka:
    image: apache/kafka:4.1.0
    hostname: kafka
    ports:
      - "9092:9092"
    environment:
      KAFKA_NODE_ID: 1
      KAFKA_PROCESS_ROLES: broker,controller
      KAFKA_LISTENERS: CONTROLLER://:9093,PLAINTEXT://:9092
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
      KAFKA_CONTROLLER_LISTENER_NAMES: CONTROLLER
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: CONTROLLER:PLAINTEXT,PLAINTEXT:PLAINTEXT
      KAFKA_CONTROLLER_QUORUM_VOTERS: 1@kafka:9093
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1
      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1

  schema-registry:
    image: confluentinc/cp-schema-registry:7.7.0
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:9092
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081

  connect:
    image: debezium/connect:3.5.0.Final
    depends_on:
      - kafka
      - schema-registry
    ports:
      - "8083:8083"
    environment:
      BOOTSTRAP_SERVERS: kafka:9092
      GROUP_ID: connect-dev
      CONFIG_STORAGE_TOPIC: connect-configs
      OFFSET_STORAGE_TOPIC: connect-offsets
      STATUS_STORAGE_TOPIC: connect-statuses
      CONFIG_STORAGE_REPLICATION_FACTOR: 1
      OFFSET_STORAGE_REPLICATION_FACTOR: 1
      STATUS_STORAGE_REPLICATION_FACTOR: 1
      KEY_CONVERTER: io.confluent.connect.avro.AvroConverter
      VALUE_CONVERTER: io.confluent.connect.avro.AvroConverter
      KEY_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      VALUE_CONVERTER_SCHEMA_REGISTRY_URL: http://schema-registry:8081
      # Enable all Debezium transforms
      CONNECT_PLUGIN_PATH: /kafka/connect,/usr/share/java

  postgres:
    image: postgres:17
    ports:
      - "5432:5432"
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
      POSTGRES_DB: inventory
    command:
      - "postgres"
      - "-c"
      - "wal_level=logical"
      - "-c"
      - "max_replication_slots=5"
      - "-c"
      - "max_wal_senders=5"
    volumes:
      - ./init-db.sql:/docker-entrypoint-initdb.d/init.sql

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    depends_on:
      - kafka
      - schema-registry
      - connect
    ports:
      - "8080:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_NAME: connect
      KAFKA_CLUSTERS_0_KAFKACONNECT_0_ADDRESS: http://connect:8083

Monitoring

JMX Metrics

Kafka Connect exposes metrics via JMX. The Prometheus JMX Exporter makes these scrapable. Key MBeans:

Worker-level metrics (kafka.connect:type=connect-worker-metrics):

connector-count — number of connectors managed by this worker
task-count — number of tasks on this worker
connector-startup-success-total, connector-startup-failure-total
task-startup-success-total, task-startup-failure-total

Task-level source metrics (kafka.connect:type=source-task-metrics,connector=*,task=*):

source-record-poll-rate — records polled per second from the source
source-record-poll-total — total records polled
source-record-write-rate — records written to Kafka per second
source-record-active-count — records currently in the in-memory queue
poll-batch-max-time-ms, poll-batch-avg-time-ms — poll batch latency

Task-level sink metrics (kafka.connect:type=sink-task-metrics,connector=*,task=*):

sink-record-read-rate — records read from Kafka per second
sink-record-send-rate — records sent to the sink per second
sink-record-active-count — records in flight (read but not yet committed)
offset-commit-completion-rate, offset-commit-max-time-ms
put-batch-max-time-ms — maximum time for a sink put() call

Error metrics (kafka.connect:type=task-error-metrics,connector=*,task=*):

total-errors-logged
total-records-skipped
deadletterqueue-produce-requests — DLQ write attempts
deadletterqueue-produce-failures — DLQ write failures (alert on this being > 0)

Prometheus JMX Exporter Config for Strimzi

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38


apiVersion: v1
kind: ConfigMap
metadata:
  name: connect-metrics-config
  namespace: kafka
data:
  metrics-config.yml: |
    lowercaseOutputName: true
    lowercaseOutputLabelNames: true
    rules:
      # Worker metrics
      - pattern: "kafka.connect<type=connect-worker-metrics><>(.+)"
        name: kafka_connect_worker_$1
        type: GAUGE

      # Source task metrics
      - pattern: "kafka.connect<type=source-task-metrics, connector=(.+), task=(.+)><>(.+)"
        name: kafka_connect_source_task_$3
        labels:
          connector: "$1"
          task: "$2"
        type: GAUGE

      # Sink task metrics
      - pattern: "kafka.connect<type=sink-task-metrics, connector=(.+), task=(.+)><>(.+)"
        name: kafka_connect_sink_task_$3
        labels:
          connector: "$1"
          task: "$2"
        type: GAUGE

      # Error metrics
      - pattern: "kafka.connect<type=task-error-metrics, connector=(.+), task=(.+)><>(.+)"
        name: kafka_connect_task_error_$3
        labels:
          connector: "$1"
          task: "$2"
        type: GAUGE

Key Alerts

Alert	Condition	Severity
Connector down	Connector state != RUNNING via REST API	Critical
Task failed	Task state == FAILED	Critical
High DLQ rate	`deadletterqueue-produce-requests` > 0 for sustained period	Warning
Source lag	`source-record-active-count` near `max.queue.size`	Warning
Sink commit slow	`offset-commit-max-time-ms` > 5000	Warning
PG slot lag	`pg_wal_lsn_diff(current, restart_lsn)` > 5 GB	Critical
DLQ write failures	`deadletterqueue-produce-failures` > 0	Critical

REST API Health Checks

1
2
3
4
5
6
7


# Quick health check — connector states
curl -s 'http://connect:8083/connectors?expand=status' | \
  jq '[to_entries[] | {name: .key, state: .value.status.connector.state, tasks: [.value.status.tasks[] | select(.state != "RUNNING")]}] | .[] | select(.state != "RUNNING" or (.tasks | length) > 0)'

# List failed tasks across all connectors
curl -s 'http://connect:8083/connectors?expand=status' | \
  jq -r '.[] | .status | .tasks[] | select(.state == "FAILED") | "\(.id) \(.trace)"'

Common Pitfalls

1. Replication Slot Lag (PostgreSQL)

Already covered above. The short version: if your Debezium connector stops for long enough, your PostgreSQL disk fills with unreclaimed WAL. Monitor pg_replication_slots continuously. Set max_slot_wal_keep_size. Use heartbeat queries on low-traffic databases.

2. Consumer Group Name Collisions

Every Kafka Connect worker cluster needs a unique group.id. If two Connect clusters share a group.id, they will join the same consumer group and fight over internal topic partitions, causing continuous rebalances. Name your clusters descriptively: connect-cdc-prod, connect-sink-prod, not connect-1.

3. Connector Class Not Found

ClassNotFoundException for a connector class usually means:

The JAR is not in any directory listed in plugin.path
The JAR is in plugin.path but its dependencies are missing (put the full connector distribution, not just the main JAR)
The JAR was added after the worker started (workers scan plugin.path at startup; restart required)
The JAR is in the Kafka libs/ directory but not isolated — this can cause classloader conflicts; keep connector JARs in their own subdirectories under plugin.path

Use the REST API to confirm what plugins are loaded: GET /connector-plugins.

4. Offset Reset Behavior

If you delete a connector and recreate it with the same name, Connect will look up the last saved offsets for that connector name in the offset topic. It will resume from where it left off, not from the beginning. To start fresh, either use a new connector name or manually delete the offset entries from the offset storage topic (requires direct Kafka tooling — there is no REST API for this).

5. Memory Pressure from Large Batch Sizes

Source connectors buffer records in a queue before writing to Kafka. The queue is bounded by max.queue.size (default 500 records) and max.queue.bytes (default 128 MB, added in Connect 2.6). On high-throughput connectors or connectors producing large messages (e.g., blob columns in Debezium), this queue can consume significant heap. Tune max.queue.size, max.queue.bytes, and worker heap (-Xmx) together.

6. Schema Evolution Failures at Runtime

If a source connector produces a record with a schema that is incompatible with the registered schema in the Schema Registry, the converter will throw an exception and the task will fail (or the record will go to DLQ if errors.tolerance=all). The most common cause is a DDL change in the source database that Debezium detects and reflects in the Avro schema, which then fails compatibility checks.

Mitigation strategies:

Use FULL_TRANSITIVE compatibility for CDC topics — it’s the strictest and safest.
Monitor the Schema Registry for new schema registrations (alert on unexpected schema changes).
Use the Debezium schema.name.adjustment.mode and explicitly manage your Schema Registry compatibility before making DDL changes.

7. Task Imbalance After Worker Loss

When a worker fails, its tasks are rebalanced to surviving workers. If the surviving workers are already near capacity, tasks may fail to start due to resource exhaustion (database connection limits, memory). Always provision workers with headroom. A rule of thumb: size your cluster to handle expected load with one worker down.

8. Strimzi: Connector Paused by Operator

If you set pause: true on a KafkaConnector resource, the operator pauses the connector. This is correct behavior. But if a connector enters PAUSED state unexpectedly, check whether the KafkaConnector spec has pause: true — someone may have edited it via GitOps while debugging and forgotten to restore it.

Kafka 4.0 and 4.1 Connect Changes

Kafka 4.0 (March 2025):

ZooKeeper fully removed — KRaft is now the only mode
Java 17 minimum for brokers, Connect, and tools
KIP-1032: Connect upgraded from JavaEE to Jakarta EE 10 APIs — connector JARs using old javax.* imports must be updated
Removed deprecated GET /connectors/{connector}/tasks-config endpoint — use GET /connectors/{connector}/tasks

Kafka 4.1 (September 2025):

KIP-891: Multiple connector plugin versions on a single worker — enables safe in-place connector upgrades
KIP-877: Plugins and connectors can now register custom JMX metrics via their task context

Version Reference

Component	Current Stable	Notes
Apache Kafka	4.1 (Sep 2025)	KRaft-only, Java 17 minimum
Debezium	3.5.0.Final (Mar 2026)	Adds CockroachDB snapshot, Oracle 23c, IBMi JDBC sink
Strimzi	0.51 (2025)	v1 stable CRD API, Kubernetes 1.30+ required
Confluent Platform	7.7.x	Kafka 3.7 based
Schema Registry (CP)	7.7.x	Bundled with Confluent Platform
Confluent Hub S3 Sink	10.5.x	Uses AWS SDK v2 (breaking from 10.x → 11.x)

Kafka Connect is the right answer for moving data into and out of Kafka at scale — when you deploy it correctly. The framework handles more than it appears to: offset management, distributed coordination, fault tolerance, schema negotiation, and error routing are all there out of the box. The connectors themselves (especially Debezium) are mature, well-tested, and trusted in production at significant scale. The operational complexity is real but manageable once you understand what the system is actually doing: using Kafka itself as the coordination and storage layer, leaving you with only the connector logic and monitoring to reason about.

The most important things to get right in production: monitor your replication slot lag from day one if you’re using Debezium on PostgreSQL, design your schema evolution strategy before you need it, size your workers with headroom, and use Strimzi on Kubernetes to make connector lifecycle a GitOps operation rather than a series of curl commands.

Sources: