Overview
Kafka Connect is the integration layer of the Kafka ecosystem — a framework for reliably moving data between Kafka and external systems (databases, filesystems, APIs) without writing producers/consumers from scratch. Debezium, built on Connect, provides Change Data Capture (CDC) from databases via binlog/WAL reading. Understanding Connect removes the need to hand-write data pipeline code for most integration use cases.
Architecture
KAFKA CONNECT ARCHITECTURE
════════════════════════════════════════════════════════
External Systems Kafka Connect Kafka
────────────────────── ─────────────────────────── ─────────
PostgreSQL ──────────────▶ Source Connector ─────────────▶ Topic
MySQL ──────────────▶ (reads external, │
REST API ──────────────▶ publishes to Kafka) │
▼
Elasticsearch ◀──────────── Sink Connector ◀─────────── Topic
S3/GCS ◀──────────── (reads from Kafka,
Snowflake ◀──────────── writes to external)
DEPLOYMENT MODES:
Standalone: single JVM process (dev, small ops)
Distributed: multiple workers, tasks auto-distributed
fault-tolerant: task reassigned on worker failure
REST API for connector management
TASK PARALLELISM:
Connector splits work into tasks.tasks.max tasks.
Each task runs independently on a worker.
JDBC source: each table = one task
Debezium: one task per connector (single binlog stream)
SMT (SINGLE MESSAGE TRANSFORM)
═════════════════════════════════════════���══════════════
Lightweight per-message transformations applied inline:
ReplaceField → rename or drop fields
InsertField → add static or metadata fields
MaskField → mask sensitive data (PII)
Filter → drop messages matching a condition
TimestampRouter → route to topic based on date
Chain multiple SMTs:
"transforms": "addTimestamp,maskPII,renameId",
"transforms.addTimestamp.type": "InsertField$Value",
"transforms.maskPII.type": "MaskField$Value",
...
Technical Implementation
Debezium PostgreSQL CDC Connector
// POST http://kafka-connect:8083/connectors
{
"name": "postgres-orders-cdc",
"config": {
"connector.class": "io.debezium.connector.postgresql.PostgresConnector",
"database.hostname": "postgres",
"database.port": "5432",
"database.user": "debezium",
"database.password": "${file:/opt/secrets:db.password}",
"database.dbname": "orders_db",
"plugin.name": "pgoutput",
"slot.name": "debezium_orders",
"table.include.list": "public.orders,public.order_items",
"topic.prefix": "cdc",
"topic.naming.strategy": "io.debezium.schema.DefaultTopicNamingStrategy",
// Creates topics: cdc.public.orders, cdc.public.order_items
"transforms": "unwrap,addSource",
"transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
"transforms.unwrap.drop.tombstones": "false",
"transforms.unwrap.delete.handling.mode": "rewrite",
// ExtractNewRecordState: unwrap Debezium envelope → just the row data
// Without: {"before": {...}, "after": {...}, "op": "u", "ts_ms": 123}
// With unwrap: {"id": 1, "status": "CONFIRMED", ...}
"transforms.addSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addSource.static.field": "source_system",
"transforms.addSource.static.value": "orders-db",
"heartbeat.interval.ms": "30000",
"snapshot.mode": "initial"
}
}
Custom Source Connector (REST API → Kafka)
// Poll a REST API and publish results to Kafka
public class RestApiSourceConnector extends SourceConnector {
private Map<String, String> config;
@Override
public void start(Map<String, String> props) {
this.config = props;
}
@Override
public Class<? extends Task> taskClass() {
return RestApiSourceTask.class;
}
@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
// For APIs that can be sharded by page/shard, create multiple tasks
return List.of(config); // single task for sequential polling
}
@Override
public ConfigDef config() {
return new ConfigDef()
.define("api.url", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "REST API URL")
.define("poll.interval.ms", ConfigDef.Type.LONG, 5000L, ConfigDef.Importance.MEDIUM, "Poll interval")
.define("topic", ConfigDef.Type.STRING, ConfigDef.Importance.HIGH, "Target Kafka topic");
}
}
public class RestApiSourceTask extends SourceTask {
private String apiUrl;
private String topic;
private long pollIntervalMs;
private String lastId;
@Override
public void start(Map<String, String> props) {
this.apiUrl = props.get("api.url");
this.topic = props.get("topic");
this.pollIntervalMs = Long.parseLong(props.get("poll.interval.ms"));
// Restore last position from offset storage (survives task restart)
Map<String, Object> stored = context.offsetStorageReader()
.offset(Map.of("api.url", apiUrl));
this.lastId = stored != null ? (String) stored.get("last_id") : "0";
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
Thread.sleep(pollIntervalMs);
List<ApiRecord> records = fetchNewRecords(apiUrl, lastId);
if (records.isEmpty()) return Collections.emptyList();
this.lastId = records.get(records.size() - 1).getId();
return records.stream()
.map(record -> new SourceRecord(
Map.of("api.url", apiUrl), // partition: source identifier
Map.of("last_id", record.getId()), // offset: resume point
topic,
Schema.STRING_SCHEMA,
record.getId(), // key
Schema.STRING_SCHEMA,
toJson(record) // value
))
.collect(toList());
}
}
JDBC Sink Connector Configuration
{
"name": "postgres-orders-sink",
"config": {
"connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
"connection.url": "jdbc:postgresql://postgres:5432/analytics_db",
"connection.user": "sink_user",
"connection.password": "${file:/opt/secrets:sink.password}",
"topics": "orders-enriched",
"table.name.format": "orders_analytics",
"insert.mode": "upsert", // INSERT ... ON CONFLICT DO UPDATE
"pk.mode": "record_key", // use Kafka message key as PK
"pk.fields": "order_id",
"auto.create": "true", // CREATE TABLE if not exists
"auto.evolve": "true", // ALTER TABLE on new fields
"batch.size": "500", // batch inserts for throughput
"max.retries": "10",
"retry.backoff.ms": "3000"
}
}
Interview Preparation
Q: What is Change Data Capture (CDC) and why use it over polling?
CDC reads the database's binary log (PostgreSQL WAL, MySQL binlog) to capture every row-level change in real-time — inserts, updates, deletes. Compared to polling (SELECT * WHERE updated_at > lastPoll): CDC captures deletes (polling can't see deleted rows), captures all intermediate states (not just latest), has near-real-time latency (milliseconds vs seconds), and doesn't add load to the database with periodic queries. CDC is more complex to set up (requires replication slots, binlog access) but is the standard approach for production data streaming from databases. Debezium is the most widely used open-source CDC tool for Kafka.
Q: What is an SMT (Single Message Transform) and when should you avoid them?
SMTs are stateless, per-message transformations applied within the Connect worker before or after routing. They're good for simple operations: renaming fields, masking PII, adding metadata, routing messages to different topics based on field values. Avoid SMTs for: complex logic (use a Kafka Streams app instead), aggregations (stateless per message — can't aggregate), transformations that need data from multiple messages, or transformations that require external lookups. SMTs that are "heavy" slow down the connector thread and reduce throughput. The rule: SMTs for minor field manipulation, Kafka Streams for business logic.
Q: How does Debezium handle schema changes in the source database?
Debezium tracks the database schema history in a special Kafka topic (schema.history.internal.kafka.topic). When a DDL change is detected in the WAL (e.g., ALTER TABLE orders ADD COLUMN discount DECIMAL), Debezium updates the schema, publishes the updated schema to the Schema Registry (if configured), and new events include the new field. Old events don't change — this requires backward-compatible schema evolution (new columns need a default or be nullable). If using Avro + Schema Registry, Debezium validates schema compatibility before publishing. Incompatible changes (rename, type change) require a connector reset and re-snapshot.
Learning Resources
- Kafka Connect Documentation
- Debezium Documentation — CDC connectors
- Confluent Hub — connector catalog
- Kafka Connect SMT Reference