Kafka Connect
Kafka Connect is a framework, included with Apache Kafka, that moves data between Kafka and other systems such as databases, file systems, cloud storage and search engines, without writing custom code.
This guide explains Connect’s components, how it works, its standalone and distributed modes, the REST API, converters, transforms and error handling, with formatted configuration and command-line examples and a custom connector written in Java. It builds on Kafka Introduction and Kafka Topic.
1. What Is Kafka Connect?
In simple words, Kafka Connect acts like a bridge. It takes data from one system and sends it to Kafka, or takes data from Kafka and sends it to another system, automatically and continuously.
MySQL, PostgreSQL, files, queues, APIs (source systems)
|
v
Kafka Connect: source connectors
|
v
Kafka topics
|
v
Kafka Connect: sink connectors
|
v
Elasticsearch, S3, data warehouse (target systems)You choose a ready-made connector, give it a configuration, and Connect handles reading, writing, scaling, progress tracking and restarts.
2. Why Do We Need Kafka Connect?
In real-world applications, data comes from many sources such as MySQL, PostgreSQL, log files or APIs, and has to reach many destinations. Writing custom code for each integration is time-consuming and error-prone, because every program must handle:
- Connecting to and reading from the source system.
- Remembering how far it has read, so nothing is lost or repeated after a restart.
- Converting data formats.
- Retries, failures and scaling.
Kafka Connect solves this problem by providing a common framework and ready-made connectors that already handle all of this.
3. Key Components of Kafka Connect
| Component | Meaning |
|---|---|
| Connector | A configured job that defines what data moves where, such as “copy table users into Kafka” |
| Source connector | Reads data from an external system and writes it into Kafka topics |
| Sink connector | Reads data from Kafka topics and writes it to an external system |
| Task | A unit of work that actually moves the data; a connector can split its work into several tasks |
| Worker | A Kafka Connect process (JVM) that runs connectors and tasks |
| Converter | Translates between Kafka’s bytes and Connect’s internal data, for example JSON or Avro |
| Transform (SMT) | A small change applied to each record as it passes through, such as renaming a field |
| Dead letter queue | A topic where a sink connector can send records it could not process |
4. Features of Kafka Connect
| Feature | Explanation |
|---|---|
| No coding required | Most integrations are done with configuration instead of code |
| Scalable | In distributed mode, tasks are spread across many workers to handle large volumes |
| Fault tolerant | If a worker stops, its connectors and tasks move to the other workers automatically. A task that fails because of an error stays failed until you restart it, usually through the REST API |
| Progress tracking | Connect saves offsets, so connectors continue where they stopped after a restart |
| Reusable connectors | Many pre-built connectors exist for databases, cloud services, storage and message systems |
| Data transformation | Single Message Transforms (SMTs) modify data while it is being moved |
| Easy management | A REST API creates, updates, pauses and monitors connectors |
5. How Kafka Connect Works
SOURCE SIDE
MySQL table
|
v
+-------------------- Connect worker --------------------+
| source task -> transforms (SMTs) -> converter |
| (saves source offsets in Kafka) |
+--------------------------------------------------------+
|
v
Kafka topic
SINK SIDE
Kafka topic
|
v
+-------------------- Connect worker --------------------+
| converter -> transforms (SMTs) -> sink task |
| (commits consumer offsets in Kafka) |
+--------------------------------------------------------+
|
v
Elasticsearch- A source task reads new data, turns it into records, and remembers its position (for example a database log position) as a source offset.
- Records pass through optional transforms, then the converter turns them into bytes for Kafka.
- A sink task is a Kafka consumer: it reads records, converts and transforms them, writes them to the target, and commits its consumer offsets.
6. Kafka Connect Example
Let’s say you have a MySQL database that stores user information, and you want to stream this data to Kafka in real time:
- A source connector reads changes from MySQL.
- The data is published into a Kafka topic.
- A sink connector reads the data from Kafka and writes it to Elasticsearch.
- The data is stored in Elasticsearch for searching and analytics.
This entire flow happens automatically without writing any custom integration code. Other applications can also read the same topic, for example to send welcome emails to new users. Section 11 shows the configuration for this example.
7. Kafka Connect Modes
| Aspect | Standalone mode | Distributed mode |
|---|---|---|
| Processes | One worker | Several workers forming a group |
| Configuration | Properties files on the command line | JSON sent to the REST API |
| Where state is kept | A local file | Kafka topics |
| Fault tolerance | None; if the process stops, data movement stops | Tasks move to other workers automatically |
| Scaling | Limited to one machine | Add workers to share the load |
| Use for | Development, testing, simple single-machine jobs | Production |
8. Quick Start: Standalone Mode
Kafka ships with simple file connectors that are perfect for learning. In Kafka 3.2 and later they are not loaded by default, so add their JAR to the plugin path first:
bootstrap.servers=localhost:9092
key.converter=org.apache.kafka.connect.storage.StringConverter
value.converter=org.apache.kafka.connect.storage.StringConverter
offset.storage.file.filename=/tmp/connect.offsets
# Folder or JAR files where connector plugins are found
plugin.path=libs/connect-file-4.0.0.jarname=local-file-source
connector.class=FileStreamSource
tasks.max=1
file=/tmp/input.txt
topic=file-lines# Start a standalone worker with the file source connector
bin/connect-standalone.sh config/connect-standalone.properties file-source.properties
# In another terminal: add lines to the file...
echo "first line" >> /tmp/input.txt
echo "second line" >> /tmp/input.txt
# ...and watch them arrive in Kafka
bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 \
--topic file-lines --from-beginningEach new line in the file becomes a message in the file-lines topic. Stop and restart the worker, and it continues from where it stopped, because it saved its position in /tmp/connect.offsets.
9. Running Kafka Connect in Distributed Mode
In production, run several workers with the same group.id. They form a Connect cluster, store their state in Kafka topics, and share the connectors’ tasks:
bootstrap.servers=broker1:9092,broker2:9092,broker3:9092
# Workers with the same group.id form one Connect cluster
group.id=connect-cluster
# Kafka topics where Connect stores its state
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
status.storage.replication.factor=3
# Default data format
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
# REST API and plugins
listeners=http://0.0.0.0:8083
plugin.path=/opt/connectors
# Allow ${file:...} references so passwords stay out of connector configs
config.providers=file
config.providers.file.class=org.apache.kafka.common.config.provider.FileConfigProviderbin/connect-distributed.sh config/connect-distributed.propertiesInstall connector plugins by placing their JAR folders in plugin.path on every worker. In distributed mode, connectors are created through the REST API, not with properties files.
10. Managing Connectors with the REST API
curl -X POST http://localhost:8083/connectors \
-H "Content-Type: application/json" \
-d '{
"name": "orders-file-sink",
"config": {
"connector.class": "FileStreamSink",
"tasks.max": "1",
"topics": "orders",
"file": "/tmp/orders-out.txt"
}
}'curl -s http://localhost:8083/connectors/orders-file-sink/status{
"name": "orders-file-sink",
"connector": { "state": "RUNNING", "worker_id": "10.0.0.5:8083" },
"tasks": [
{ "id": 0, "state": "RUNNING", "worker_id": "10.0.0.6:8083" }
],
"type": "sink"
}| Request | Purpose |
|---|---|
GET /connectors | List all connectors |
GET /connector-plugins | List installed connector plugins |
PUT /connectors/{name}/config | Create or update a connector’s configuration |
GET /connectors/{name}/status | Show connector and task states |
PUT /connectors/{name}/pause and /resume | Temporarily stop and continue data movement |
POST /connectors/{name}/restart?includeTasks=true&onlyFailed=true | Restart failed tasks |
DELETE /connectors/{name} | Remove a connector |
Because PUT /connectors/{name}/config both creates and updates, it is convenient for deployment scripts that keep connector configurations in version control.
11. Real Example: Database to Kafka to Search Engine
This example implements the MySQL to Elasticsearch flow from section 6 with two widely used connector plugins: a change data capture (CDC) source connector that reads every insert, update and delete from MySQL’s change log, and an Elasticsearch sink connector. Both are installed separately into plugin.path.
{
"name": "mysql-users-source",
"config": {
"connector.class": "io.debezium.connector.mysql.MySqlConnector",
"tasks.max": "1",
"database.hostname": "mysql",
"database.port": "3306",
"database.user": "connect_user",
"database.password": "${file:/opt/secrets/mysql.properties:password}",
"database.server.id": "5401",
"topic.prefix": "shop",
"table.include.list": "shop.users",
"schema.history.internal.kafka.bootstrap.servers": "broker1:9092",
"schema.history.internal.kafka.topic": "schema-history.shop"
}
}{
"name": "users-search-sink",
"config": {
"connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
"tasks.max": "2",
"topics": "shop.shop.users",
"connection.url": "http://elasticsearch:9200",
"key.ignore": "false",
"schema.ignore": "true"
}
}curl -X POST -H "Content-Type: application/json" \
--data @mysql-users-source.json http://localhost:8083/connectors
curl -X POST -H "Content-Type: application/json" \
--data @users-search-sink.json http://localhost:8083/connectorsThe source connector writes changes for table users to the topic shop.shop.users (prefix, database, table). The password is read from a separate file through the file config provider, so it never appears in the REST API. Check each plugin’s documentation and license for the exact options of your version; the CDC source also needs a MySQL user with replication permissions and binary logging enabled.
12. Converters and Data Formats
| Converter | Format in Kafka | Notes |
|---|---|---|
StringConverter | Plain text | Simple, no structure |
JsonConverter | JSON | With schemas.enable=true, each message also carries its schema, which makes messages larger |
ByteArrayConverter | Raw bytes | Passes data through unchanged |
| Avro, Protobuf or JSON Schema converters | Compact binary or JSON with a schema registry | Third-party converters; schemas are stored once in a registry and evolve safely |
"key.converter": "org.apache.kafka.connect.storage.StringConverter",
"value.converter": "org.apache.kafka.connect.json.JsonConverter",
"value.converter.schemas.enable": "false"Converter settings in a connector override the worker defaults. The producer and consumer sides of a pipeline must agree on the format, or the sink will fail to read the data.
13. Single Message Transforms (SMTs)
Single Message Transforms make small, record-by-record changes without writing code. Several transforms can be chained:
"transforms": "addSource,route",
"transforms.addSource.type": "org.apache.kafka.connect.transforms.InsertField$Value",
"transforms.addSource.static.field": "source_system",
"transforms.addSource.static.value": "mysql-shop",
"transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
"transforms.route.regex": "shop\\.shop\\.(.*)",
"transforms.route.replacement": "shop-$1"Here every record gets a source_system field, and the topic shop.shop.users is renamed to shop-users.
| Built-in transform | What it does |
|---|---|
InsertField | Adds a fixed value, topic name, partition, offset or timestamp as a field |
ReplaceField | Keeps, drops or renames fields |
MaskField | Replaces sensitive field values, for example with an empty value |
ValueToKey and ExtractField | Build the record key from value fields |
RegexRouter and TimestampRouter | Change the topic name |
Filter with predicates | Drop records that match a condition |
Cast | Change field data types |
Use SMTs for simple changes. For joins, aggregations or complex logic, use a stream processing application instead.
14. Error Handling and Dead Letter Queues
By default (errors.tolerance=none), a task stops on the first record it cannot convert or transform. For sink connectors you can instead skip bad records and keep them in a dead letter queue (DLQ) topic:
"errors.tolerance": "all",
"errors.retry.timeout": "300000",
"errors.log.enable": "true",
"errors.log.include.messages": "false",
"errors.deadletterqueue.topic.name": "dlq-users-search-sink",
"errors.deadletterqueue.topic.replication.factor": "3",
"errors.deadletterqueue.context.headers.enable": "true"- Failed records go to
dlq-users-search-sink, with headers describing the error. - Temporary errors are retried for up to 5 minutes before the record counts as failed.
- Watch the DLQ topic: a growing DLQ means data is not reaching the target.
Only sink connectors support a DLQ. If a task fails anyway, find the cause in the status API or worker logs, fix it, and restart the task through the REST API.
15. Writing a Custom Source Connector in Java
If no ready-made connector fits, you can write your own. A connector has two parts: the connector class (configuration and splitting work into tasks) and the task class (moving data). This small source connector writes an increasing counter to a topic and resumes from the last value after a restart:
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>connect-api</artifactId>
<version>4.0.0</version>
<scope>provided</scope>
</dependency>package com.example.connect;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.source.SourceConnector;
public class CounterSourceConnector extends SourceConnector {
static final String TOPIC = "topic";
static final String INTERVAL_MS = "interval.ms";
private static final ConfigDef CONFIG = new ConfigDef()
.define(TOPIC, ConfigDef.Type.STRING, ConfigDef.Importance.HIGH,
"Topic to write counter values to")
.define(INTERVAL_MS, ConfigDef.Type.LONG, 1000L, ConfigDef.Importance.MEDIUM,
"Pause between values in milliseconds");
private Map<String, String> settings;
@Override
public void start(Map<String, String> props) {
this.settings = props;
}
@Override
public Class<? extends Task> taskClass() {
return CounterSourceTask.class;
}
@Override
public List<Map<String, String>> taskConfigs(int maxTasks) {
return List.of(settings); // one counter, so one task is enough
}
@Override
public void stop() {
}
@Override
public ConfigDef config() {
return CONFIG;
}
@Override
public String version() {
return "1.0.0";
}
}package com.example.connect;
import java.util.List;
import java.util.Map;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
public class CounterSourceTask extends SourceTask {
private static final Map<String, String> SOURCE_PARTITION = Map.of("source", "counter");
private String topic;
private long intervalMs;
private long next;
@Override
public void start(Map<String, String> props) {
topic = props.get(CounterSourceConnector.TOPIC);
intervalMs = Long.parseLong(props.getOrDefault(CounterSourceConnector.INTERVAL_MS, "1000"));
// Resume from the last saved position, if any
Map<String, Object> saved = context.offsetStorageReader().offset(SOURCE_PARTITION);
next = (saved == null) ? 0L : (Long) saved.get("position") + 1;
}
@Override
public List<SourceRecord> poll() throws InterruptedException {
Thread.sleep(intervalMs);
SourceRecord record = new SourceRecord(
SOURCE_PARTITION,
Map.of("position", next), // source offset saved by Connect
topic,
Schema.INT64_SCHEMA,
next);
next++;
return List.of(record);
}
@Override
public void stop() {
}
@Override
public String version() {
return "1.0.0";
}
}Package the classes as a JAR, copy it into a folder listed in plugin.path on every worker, restart the workers, and create the connector with "connector.class": "com.example.connect.CounterSourceConnector" and a topic setting. Connect takes care of producing the records, saving offsets and restarting the task on another worker if one fails.
16. Monitoring Kafka Connect
| What to watch | How |
|---|---|
| Connector and task states | GET /connectors/{name}/status; alert on any FAILED state |
| Sink connector lag | Sink tasks are consumers in the group connect-<connector name>; check lag with kafka-consumer-groups.sh |
| Throughput and errors | JMX metrics such as records read and written per task, and error counts |
| Dead letter queues | Message count in each DLQ topic |
| Worker logs | Stack traces explaining why a task failed |
17. Kafka Connect vs Writing Your Own Code
| Aspect | Kafka Connect | Custom producer or consumer code |
|---|---|---|
| Effort | Configuration of an existing connector | Write, test and maintain code |
| Offset tracking and restarts | Built in | You implement it |
| Scaling | Add workers or tasks | You design it |
| Data formats | Pluggable converters | You implement serialization |
| Best for | Moving data between standard systems | Business logic and custom processing |
Use Connect to move data, and normal applications or stream processors to process it.
18. Advantages of Kafka Connect
- Reduces development effort.
- Provides a standard way to move data in and out of Kafka.
- Handles offsets, retries and worker failures automatically.
- Is easy to configure and manage through a REST API.
- Has a large ecosystem of ready-made connectors.
19. Best Practices
| Do | Avoid |
|---|---|
| Use distributed mode in production with at least two workers | Running production pipelines in standalone mode |
| Use replication factor 3 for Connect’s internal topics | Low replication for config, offset and status topics |
| Keep passwords in config providers or secret stores | Putting passwords directly in connector JSON |
Store connector configs in version control and deploy with PUT .../config | Creating connectors by hand without a record |
Set tasks.max based on partitions or tables | Very high task counts that only add overhead |
| Use a DLQ for sink connectors and monitor it | Silently dropping bad records |
| Alert on failed tasks and restart them after fixing the cause | Assuming failed tasks restart by themselves |
| Check each connector plugin’s license and version compatibility | Mixing plugin versions across workers |
20. Summary
Kafka Connect is a powerful and easy-to-use framework that simplifies data integration with Kafka. Source connectors bring data in, sink connectors send it out, and workers run their tasks with built-in offset tracking, scaling and failover. Converters handle data formats, SMTs make small changes on the way, and dead letter queues catch bad records, all managed through a REST API.
It allows teams to focus on business logic instead of writing and maintaining complex data pipelines. With ready-made connectors, scalability and fault tolerance, Kafka Connect is an essential tool for building reliable real-time data systems.
Frequently Asked Questions (FAQ)
Q1. What is Kafka Connect in simple words?
Kafka Connect is a framework that moves data between Kafka and other systems using ready-made connectors. You configure a connector instead of writing code, and Connect handles reading, writing, progress tracking, scaling and failover.
Q2. Is Kafka Connect part of Apache Kafka?
Yes. The Connect framework, its REST API, the standalone and distributed workers and a few simple connectors ship with Apache Kafka. Most real connectors, for databases, cloud storage or search engines, are separate plugins that you install.
Q3. What is the difference between a source and a sink connector?
A source connector reads from an external system, such as a database, and writes into Kafka topics. A sink connector reads from Kafka topics and writes to an external system, such as a search engine or data warehouse.
Q4. What is the difference between a connector and a task?
A connector is the configured job and decides how to split the work. Tasks do the actual data copying. For example, a sink connector reading a topic with 6 partitions could run up to 6 tasks in parallel, set by tasks.max.
Q5. What is a Kafka Connect worker?
A worker is a Java process that runs connectors and tasks. In standalone mode there is one worker; in distributed mode, several workers with the same group.id form a cluster and share the work.
Q6. Should I use standalone or distributed mode?
Use standalone for development, testing or simple jobs on one machine. Use distributed for production: it stores state in Kafka, moves tasks to other workers if one fails, and lets you add workers to scale.
Q7. Does Kafka Connect restart failed tasks automatically?
Not when the task itself fails with an error; it stays in the FAILED state. Fix the cause and restart it:
curl -X POST "http://localhost:8083/connectors/my-connector/restart?includeTasks=true&onlyFailed=true"What Connect does automatically is move tasks to other workers when a worker process stops, and retry temporary errors when configured.
Q8. Where does Kafka Connect store its state?
In distributed mode, in three Kafka topics: connector configurations (config.storage.topic), source offsets (offset.storage.topic) and connector and task status (status.storage.topic). Sink connectors store their progress as normal consumer group offsets. Standalone mode keeps source offsets in a local file.
Q9. How do I install a connector plugin?
Download the plugin (usually a folder of JAR files), place it in a directory listed in plugin.path on every worker, and restart the workers. Check that it appears in GET /connector-plugins.
Q10. What is a converter?
A converter translates between the bytes stored in Kafka and Connect’s internal data format. Common choices are StringConverter, JsonConverter and schema-registry-based converters for Avro, Protobuf or JSON Schema. Source and sink sides must use compatible converters.
Q11. What are Single Message Transforms (SMTs)?
Small, configurable changes applied to each record as it passes through a connector, such as adding a field, removing or masking a field, changing the key or renaming the topic. They need no code, but are meant for simple changes only.
Q12. What is a dead letter queue in Kafka Connect?
A Kafka topic where a sink connector sends records it cannot convert or transform, instead of stopping. Enable it with errors.tolerance=all and errors.deadletterqueue.topic.name, and add error details as headers with errors.deadletterqueue.context.headers.enable=true.
Q13. How do I keep passwords out of connector configurations?
Enable a config provider on the workers, for example the built-in file provider, and refer to secrets instead of writing them directly:
"database.password": "${file:/opt/secrets/mysql.properties:password}"The value is resolved by the worker at runtime and is not shown by the REST API.
Q14. How can I stream database changes into Kafka?
Use a change data capture (CDC) source connector. It reads the database’s change log and publishes every insert, update and delete as an event, usually one topic per table. Query-based JDBC source connectors are an alternative, but they cannot capture deletes and need a column to detect new rows.
Q15. How do I scale Kafka Connect?
Increase tasks.max so a connector can run more tasks in parallel, and add workers so the tasks are spread over more machines. For sink connectors, parallelism is limited by the number of topic partitions.
Q16. How do I check if a connector is working?
Call GET /connectors/{name}/status; the connector and every task should be RUNNING. For sink connectors, also check the lag of the consumer group connect-{name} and the size of any dead letter queue topic.
Q17. Can Kafka Connect transform or join data?
It can make simple per-record changes with SMTs. For filtering with complex logic, joining streams, aggregating or enriching data, use a stream processing application such as Kafka Streams between the source and sink connectors.
Q18. Can I write my own connector?
Yes. Add the connect-api dependency, implement a SourceConnector and SourceTask (or the sink equivalents), package them as a JAR, and place it in plugin.path. Connect then handles offsets, scaling and failover for you.
Q19. How do I update a running connector?
Send the full new configuration with PUT /connectors/{name}/config. Connect saves it and restarts the connector’s tasks with the new settings. The same request creates the connector if it does not exist yet.
Q20. Is Kafka Connect related to Spring Boot?
Not directly. Kafka Connect runs as its own set of worker processes, separate from your applications. A Spring Boot application usually consumes the topics that a source connector fills, or produces to topics that a sink connector reads.