Kafka Connect

Intermediate

Kafka Connect

Kafka Connect is a scalable framework for streaming data between Kafka and external systems (databases, file systems, cloud services) without writing custom producer/consumer code.

Overview

Kafka Connect is a distributed, fault-tolerant runtime for moving data between Kafka and external systems. Instead of writing custom producer or consumer code for each integration, you deploy connector plugins — source connectors pull data into Kafka from external systems (databases, message queues, files), and sink connectors push Kafka data out to external systems (Elasticsearch, S3, JDBC, Snowflake). Connect runs as a cluster of worker processes that distribute and rebalance connector tasks automatically. Configuration is JSON-based and submitted via REST API — no code is needed for most integrations. Connect handles offset tracking, error handling, schema registry integration, and exactly-once delivery, making it the standard data ingestion layer in the Kafka ecosystem.

Connect cluster setup and REST API

A distributed Kafka Connect cluster runs as worker processes. Connectors are created, monitored, and managed via the REST API on port 8083.

Properties + Shell — Connect distributed config + REST

# connect-distributed.properties
bootstrap.servers=kafka-1:9092,kafka-2:9092,kafka-3:9092
group.id=connect-cluster-1

# Internal topics for offset, config, and status storage
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=3
offset.storage.replication.factor=3

# Converters — Avro with Schema Registry
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

plugin.path=/usr/share/confluent-hub-components

# Start worker
bin/connect-distributed.sh config/connect-distributed.properties

# REST API — list running connectors
curl http://connect:8083/connectors

# REST API — check connector status
curl http://connect:8083/connectors/my-connector/status

Creating and monitoring a connector via REST

Connectors are deployed by POSTing a JSON config to the REST API. The Connect worker manages task lifecycle, failure recovery, and rebalancing.

Shell — JDBC Source connector REST CRUD

# Create a JDBC Source connector (polls a database table)
curl -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-jdbc-source",
    "config": {
      "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
      "connection.url": "jdbc:mysql://db:3306/orders",
      "connection.user": "connect_user",
      "connection.password": "secret",
      "mode": "timestamp+incrementing",
      "timestamp.column.name": "updated_at",
      "incrementing.column.name": "id",
      "table.whitelist": "orders",
      "topic.prefix": "mysql.",
      "poll.interval.ms": "5000",
      "tasks.max": "3"
    }
  }'

# Monitor connector
curl http://connect:8083/connectors/orders-jdbc-source/status
# {"name":"orders-jdbc-source","connector":{"state":"RUNNING"},"tasks":[...]}

# Pause / resume
curl -X PUT http://connect:8083/connectors/orders-jdbc-source/pause
curl -X PUT http://connect:8083/connectors/orders-jdbc-source/resume

# Restart a failed task
curl -X POST http://connect:8083/connectors/orders-jdbc-source/tasks/0/restart

Dead Letter Queue and error handling

Configure a DLQ to capture records that fail to deserialise or process. Without DLQ, a bad record halts the entire connector task.

Shell — S3 Sink connector with DLQ config

# S3 Sink connector with DLQ for failed records
curl -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d '{
    "name": "orders-s3-sink",
    "config": {
      "connector.class": "io.confluent.connect.s3.S3SinkConnector",
      "tasks.max": "2",
      "topics": "mysql.orders",
      "s3.region": "us-east-1",
      "s3.bucket.name": "data-lake-orders",
      "s3.part.size": "5242880",
      "flush.size": "1000",
      "storage.class": "io.confluent.connect.s3.storage.S3Storage",
      "format.class": "io.confluent.connect.s3.format.parquet.ParquetFormat",

      "errors.tolerance": "all",
      "errors.log.enable": "true",
      "errors.log.include.messages": "true",
      "errors.deadletterqueue.topic.name": "orders-dlq",
      "errors.deadletterqueue.topic.replication.factor": "3",
      "errors.deadletterqueue.context.headers.enable": "true"
    }
  }'

Key Points to Remember

1Connect eliminates custom producer/consumer code for standard integrations — use it before writing code.
2Distributed mode is production-standard; standalone mode is for single-worker development only.
3Connectors run tasks that are distributed and rebalanced across workers automatically on failure.
4Always configure a DLQ (errors.deadletterqueue.topic.name) so a bad record does not halt the connector.
5Internal topics (connect-configs, connect-offsets, connect-status) must have replication.factor=3 for HA.
6Use Confluent Hub to find pre-built connectors (jdbc, s3, elasticsearch, debezium) before writing custom plugins.

Interview Questions

What is the difference between a Kafka Connect source connector and a sink connector?

EasyConfluent

How does Kafka Connect distribute connector tasks across worker processes?

MediumLinkedIn

What happens to a connector task when a record fails to process and how do you handle it?

MediumNetflix

How does Kafka Connect track which records it has already processed after a restart?

HardUber

When would you build a custom connector plugin vs using an existing one from Confluent Hub?

MediumAmazon

Ask Aria about Kafka Connect

Your personal AI tutor — ask anything about this concept

Revision Status

Personal Notes

Discussion

Loading discussion…

Interactive Queries

Source Connectors & Debezium CDC