connect

Active contributors: Chris Egerton, Mickael Maison, Ken Huang, Greg Harris, PoAn Yang

Purpose

Kafka Connect is a framework for moving data into Kafka (source connectors) and out of Kafka (sink connectors). Connect itself is a worker process that hosts pluggable connectors, manages their tasks, distributes work across multiple workers in a cluster, and exposes a REST API. Bundled in this repo are the framework (connect/runtime/), the public API connectors implement (connect/api/), JSON converters, MirrorMaker 2 (cross-cluster replication), file-based sample connectors, and a basic-auth REST extension.

Directory layout

connect/
├── api/                    ← public SPI: SourceConnector, SinkConnector, Task, ConfigDef, etc.
├── runtime/                ← Connect framework (workers, herders, REST, plugin loading)
│   └── src/main/java/org/apache/kafka/connect/runtime/
│       ├── Connect.java
│       ├── Worker.java                          ← (~3.4K lines) per-task lifecycle owner
│       ├── WorkerSourceTask.java / WorkerSinkTask.java / ExactlyOnceWorkerSourceTask.java
│       ├── Herder.java + AbstractHerder.java   ← assignment / leader election
│       ├── distributed/                         ← DistributedHerder, DistributedConfig, KafkaConfigBackingStore, ...
│       ├── standalone/                          ← StandaloneHerder
│       ├── isolation/                           ← Plugins (classloader isolation per plugin)
│       ├── rest/                                ← REST resources (Jersey)
│       ├── errors/                              ← ErrorReporters, dead-letter queue, retries
│       └── health/
├── json/                   ← JSON Converter implementation
├── basic-auth-extension/   ← REST extension that adds HTTP Basic auth
├── transforms/             ← Single-message transforms (SMTs): Cast, ExtractField, RegexRouter, ...
├── file/                   ← FileStreamSourceConnector / SinkConnector (samples)
├── mirror/                 ← MirrorMaker 2 (replication-as-Connect)
├── mirror-client/          ← client utilities for MM2 (offset translation, RemoteClusterUtils)
└── test-plugins/           ← test fixtures

Key abstractions

Type	File	Purpose
SourceConnector / SinkConnector	connect/api/src/main/java/org/apache/kafka/connect/source/SourceConnector.java (and sink)	Public abstract classes; users extend these.
SourceTask / SinkTask	connect/api/.../source/SourceTask.java, sink/SinkTask.java	The actual record-pump unit; one connector spawns N tasks.
Worker	connect/runtime/src/main/java/org/apache/kafka/connect/runtime/Worker.java	Hosts running connector and task instances on a single JVM; owns producers / consumers per task.
Herder / AbstractHerder	connect/runtime/.../runtime/Herder.java, AbstractHerder.java	Coordinates connector / task assignment across workers in distributed mode.
DistributedHerder	connect/runtime/.../distributed/DistributedHerder.java	Distributed-mode herder using Kafka group coordination + the Connect "config topic".
StandaloneHerder	connect/runtime/.../standalone/StandaloneHerder.java	Single-process herder, configured via files; useful for development.
KafkaConfigBackingStore / KafkaOffsetBackingStore / KafkaStatusBackingStore	connect/runtime/.../storage/	Three internal Kafka topics: connect-configs, connect-offsets, connect-status.
Plugins	connect/runtime/.../isolation/Plugins.java	Connector plugin loader; gives every plugin its own classloader (isolation from the framework).
Transformation (SMT) / ConnectorClientConfigOverridePolicy	connect/api/.../transforms/Transformation.java	Interfaces for transforms and policy plugins.
ConnectMetrics	connect/runtime/.../runtime/ConnectMetrics.java	Per-worker / per-task metrics registry.
RestServer, *Resource classes	connect/runtime/.../rest/	Jersey-based REST endpoints (/connectors, /connector-plugins, /connectors/{name}/config, ...).

Distributed mode

graph TD
    subgraph Cluster
      W1[Worker A] -->|join group| GC[Group coordinator<br/>__connect-configs / __connect-status]
      W2[Worker B] --> GC
      W3[Worker C] --> GC
    end
    GC -->|elect| LD[Leader worker]
    LD -->|writes assignment| CS[(connect-configs topic)]
    CS --> W1 & W2 & W3
    W1 --> T1[SourceTask 1]
    W2 --> T2[SourceTask 2]
    W3 --> T3[SinkTask 3]

Workers form a Connect group (separate from any consumer group) using the same group coordinator on the broker. The leader writes connector + task assignment records into the connect-configs Kafka topic; followers pick them up and start the assigned tasks. Connector configs, current offsets, and status messages each live on their own internal Kafka topic — Connect is therefore a Kafka client, not a separate service with its own database.

The new "incremental cooperative rebalance" protocol minimizes task disruption when workers join / leave; it is implemented in IncrementalCooperativeAssignor.java.

Standalone mode

StandaloneHerder runs the same code paths but without distribution. Connector configs are loaded from files; offsets are stored on disk. Useful for development, copying small datasets, and the bundled file connectors. Not recommended for production multi-worker setups.

Source connector lifecycle

sequenceDiagram
    participant H as Herder
    participant W as Worker
    participant T as WorkerSourceTask
    participant Conn as User SourceConnector / Task
    participant K as KafkaProducer
    H->>W: startTask(taskId, config)
    W->>T: new WorkerSourceTask
    T->>Conn: SourceTask.initialize / start
    loop poll loop
      T->>Conn: poll() — returns List<SourceRecord>
      T->>T: apply transformations
      T->>K: send() each record
      K-->>T: ack
      T->>Conn: commitRecord / commit
    end
    H->>W: stopTask(taskId)
    W->>Conn: SourceTask.stop

Sink tasks are symmetric: a WorkerSinkTask owns a KafkaConsumer, polls records, applies transformations, and feeds batches into SinkTask.put. SinkTask.preCommit controls offset commit semantics.

Exactly-once for source connectors (KIP-618)

ExactlyOnceWorkerSourceTask wraps a transactional producer per task. Each polled batch is sent inside a transaction along with offset records, so a crash never produces duplicates downstream. Sink-side EOS depends on the destination system; Connect provides hooks (SinkTask.preCommit) for connectors to integrate.

REST API

Standard endpoints (Jersey resources under connect/runtime/.../rest/resources/):

• GET /connectors — list connectors.
• POST /connectors — create.
• GET /connectors/{name} — describe.
• PUT /connectors/{name}/config — update config.
• DELETE /connectors/{name} — delete.
• GET /connectors/{name}/status — current state of the connector and its tasks.
• POST /connectors/{name}/restart, /connectors/{name}/tasks/{taskId}/restart.
• GET /connector-plugins — installed plugins.

The REST server is a Jetty + Jersey stack. The basic-auth-extension/ module provides HTTP Basic auth via the ConfigProvider SPI.

MirrorMaker 2 (connect/mirror/)

MirrorMaker 2 is implemented as a set of source connectors (MirrorSourceConnector, MirrorCheckpointConnector, MirrorHeartbeatConnector) plus utilities. A single Connect cluster (or a dedicated MM2 daemon) replicates topics, consumer-group offsets, and ACLs between Kafka clusters. connect/mirror-client/ provides client-side helpers like RemoteClusterUtils.translateOffsets for fail-over scenarios.

Plugin isolation

Plugins (connect/runtime/.../isolation/Plugins.java) loads each connector plugin with its own PluginClassLoader. Connector JARs go under one of the directories listed in plugin.path; symbols from the framework are filtered out from the plugin classloader so that two plugins with the same dependency name can coexist with different versions.

Configuration

Top-level worker configs (in connect/runtime/.../runtime/distributed/DistributedConfig.java and WorkerConfig.java):

• bootstrap.servers
• group.id — Connect group, distinct from any consumer-group ID
• key.converter, value.converter — serializers (e.g., org.apache.kafka.connect.json.JsonConverter)
• config.storage.topic, offset.storage.topic, status.storage.topic — internal topics
• plugin.path — directories scanned for connector JARs
• rest.host.name, rest.port
• exactly.once.source.support — disabled, preparing, enabled

Connector-level configs are declared via the connector's ConfigDef (returned from Connector.config()) and validated via Connector.validate(...).

Entry points for modification

• Add a connector plugin: implement SourceConnector + SourceTask (or SinkConnector + SinkTask) in your own JAR; drop on plugin.path.
• Change rebalance / assignment: connect/runtime/.../distributed/IncrementalCooperativeAssignor.java.
• Change EOS semantics for source: connect/runtime/.../runtime/ExactlyOnceWorkerSourceTask.java.
• Add a REST endpoint: a Jersey *Resource class under connect/runtime/.../rest/resources/. Public-API changes need a KIP.
• Add or change a converter: implement Converter (e.g., look at connect/json/).

Related pages

• Modules: clients — Connect runs on top of the standard producer/consumer/admin.
• Modules: coordinators — distributed Connect uses the group coordinator.
• Features: Exactly-once semantics.