Kafka Connect & Data Integration

Kafka Connect moves data between Kafka and other systems declaratively. You post JSON configuration to a REST API; the framework handles parallelism, offset tracking, restarts and rebalancing.

The shape#

Change data capture with Debezium#

{
  "name": "orders-cdc",
  "config": {
    "connector.class": "io.debezium.connector.postgresql.PostgresConnector",
    "database.hostname": "postgres.internal",
    "database.dbname": "orders",
    "database.user": "debezium",
    "plugin.name": "pgoutput",
    "topic.prefix": "acme",
    "table.include.list": "public.outbox",
    "slot.name": "orders_slot",
    "publication.autocreate.mode": "filtered",
    "snapshot.mode": "initial",
    "transforms": "outbox",
    "transforms.outbox.type": "io.debezium.transforms.outbox.EventRouter",
    "transforms.outbox.route.by.field": "aggregate_type",
    "transforms.outbox.table.field.event.key": "aggregate_id",
    "transforms.outbox.table.field.event.payload": "payload",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "connect-dlq",
    "errors.deadletterqueue.context.headers.enable": true
  }
}

This configuration is the outbox pattern with no relay code at all. The application writes to an outbox table inside its business transaction; Debezium tails the write-ahead log, sees the insert, and the EventRouter transform routes it to a topic named from aggregate_type with aggregate_id as the key. Atomic writes, correct partitioning, and no polling loop to operate.

Two operational cautions specific to Postgres. A replication slot retains WAL until the consumer advances, so a stopped connector will grow the database's disk until it fills — monitor slot lag as carefully as consumer lag. And an initial snapshot of a large table can take hours and holds locks briefly; plan the first start for a quiet window.

A third concerns the outbox table itself, because nothing in the configuration above ever deletes from it. The rows only need to exist long enough to reach the write-ahead log, so the neatest pattern is to insert and then delete within the same transaction: Postgres writes both operations to the log, Debezium still sees the insert and routes it, and the table stays permanently empty. If that reads as too clever for the team who will maintain it, a scheduled purge of rows older than a day is equally correct. What does not work is leaving it to grow, which turns a well-designed integration into a slow table scan eighteen months later.

Sinks#

{
  "name": "orders-to-elasticsearch",
  "config": {
    "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
    "topics": "acme.public.orders",
    "connection.url": "https://es.internal:9200",
    "tasks.max": "3",
    "key.ignore": "false",
    "write.method": "upsert",
    "behavior.on.null.values": "delete",
    "batch.size": "2000",
    "linger.ms": "1000",
    "max.retries": "5",
    "retry.backoff.ms": "1000",
    "errors.tolerance": "all",
    "errors.deadletterqueue.topic.name": "es-sink-dlq"
  }
}

behavior.on.null.values: delete maps a Kafka tombstone onto an Elasticsearch delete, which keeps the index consistent with a compacted topic. Without it, deleted records linger in the index forever.

tasks.max should not exceed the topic's partition count — extra tasks sit idle, exactly as extra consumers in a group would.

Single Message Transforms#

Transforms run per record, chained in order:

{
  "transforms": "unwrap,mask,route,timestamp",
 
  "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
  "transforms.unwrap.drop.tombstones": "false",
 
  "transforms.mask.type": "org.apache.kafka.connect.transforms.MaskField$Value",
  "transforms.mask.fields": "email,phone",
  "transforms.mask.replacement": "***",
 
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "acme\\.public\\.(.*)",
  "transforms.route.replacement": "$1",
 
  "transforms.timestamp.type": "org.apache.kafka.connect.transforms.TimestampConverter$Value",
  "transforms.timestamp.field": "created_at",
  "transforms.timestamp.target.type": "Timestamp"
}

MaskField is the one worth adopting as policy. Redacting personal data at the connector means it never enters Kafka at all, so it never reaches the dozen downstream systems that consume from there. Filtering it later means it already exists in every one of them.

Keep chains short. Five transforms per record at high throughput is measurable CPU, and a long chain becomes configuration nobody can follow. Anything requiring state or more than trivial logic belongs in Kafka Streams.

Running Connect#

bootstrap.servers: kafka-1:9092,kafka-2:9092
group.id: connect-cluster
# Workers are stateless; all state lives in these three topics.
config.storage.topic: connect-configs
config.storage.replication.factor: 3
offset.storage.topic: connect-offsets
offset.storage.partitions: 25
offset.storage.replication.factor: 3
status.storage.topic: connect-status
status.storage.replication.factor: 3
key.converter: org.apache.kafka.connect.storage.StringConverter
value.converter: io.confluent.connect.avro.AvroConverter
value.converter.schema.registry.url: http://schema-registry:8081

Because state lives in Kafka, a worker is disposable — kill one and its tasks rebalance onto the others. That makes Connect straightforward to run on Kubernetes with a plain Deployment.

Manage connectors through the REST API, and keep the JSON in version control so configuration is reviewable and reproducible:

curl -s localhost:8083/connectors | jq
curl -s localhost:8083/connectors/orders-cdc/status | jq
curl -X POST localhost:8083/connectors/orders-cdc/restart?includeTasks=true

Monitoring#

Three checks catch nearly everything. Task status — a task can fail while the connector reports running, so check task state rather than connector state. Source lag, which for Debezium means replication slot lag and for a polling connector means how far behind the query is. DLQ depth, because errors.tolerance: all means bad records are skipped silently unless somebody watches where they went.

Set an alert on any task in FAILED state. Connect does not restart failed tasks automatically by default, so a failure at 2am stays failed until someone posts a restart.

What to take away#

Use Connect instead of writing integration code — it handles offsets, parallelism and restarts for free. Pair Debezium with an outbox table for atomic event publishing with a schema you control. Keep transforms to per-record cleanup, always configure a dead-letter queue, and alert on failed tasks rather than trusting connector state.