Kafka Connect Architecture and Deployment Modes

Apache Kafka Connect is a framework for streaming data between Kafka and external systems using reusable connector plugins. For platform and site reliability engineers, understanding its runtime model and deployment topologies is the difference between a Connect cluster that self-heals and one that silently stops processing. This page dissects the worker/task/connector model, the trade-offs between standalone and distributed modes, and concrete guidance for running Connect in production. Connect handles the mechanics of data movement; it is most useful when paired with a schema management layer, as explored in the broader Schema Registry & Connect Ecosystem.

Internal Architecture: Workers, Tasks, and Connectors

A Kafka Connect cluster is a group of worker processes that execute connector logic. The distinction between the logical components—connectors and tasks—and the physical runtime—workers—is central to capacity planning and troubleshooting.

A connector is a logical job that defines what data to copy and where. It is created by submitting a JSON configuration to the cluster’s REST API. A connector performs no data movement itself; it generates a set of tasks, which are the units of parallelism that actually copy data. A source connector’s tasks poll an external system and produce records to Kafka; a sink connector’s tasks consume records from Kafka and write them to an external system.

A worker is a single JVM process that runs one or more tasks. Workers own the lifecycle of tasks—starting, stopping, and reassigning them—and also serve the REST API, manage offsets, and persist configuration. Adding workers increases the pool of resources available to execute tasks, which is what gives distributed Connect its horizontal scalability.

The following diagram illustrates these components in a distributed cluster:

┌─────────────────────────────────────────────────────────────┐
│                    Kafka Connect Cluster                     │
│                                                              │
│  ┌──────────────┐  ┌──────────────┐  ┌──────────────┐       │
│  │   Worker 1   │  │   Worker 2   │  │   Worker 3   │       │
│  │              │  │              │  │              │       │
│  │ ┌──────────┐ │  │ ┌──────────┐ │  │ ┌──────────┐ │       │
│  │ │ Task A-0 │ │  │ │ Task A-1 │ │  │ │ Task B-0 │ │       │
│  │ └──────────┘ │  │ └──────────┘ │  │ └──────────┘ │       │
│  │ ┌──────────┐ │  │              │  │ ┌──────────┐ │       │
│  │ │ Task B-1 │ │  │              │  │ │ Task C-0 │ │       │
│  │ └──────────┘ │  │              │  │ └──────────┘ │       │
│  └──────────────┘  └──────────────┘  └──────────────┘       │
│                                                              │
│  Connector A (tasks.max=2)    Connector B (tasks.max=2)     │
│  Connector C (tasks.max=1)                                   │
└─────────────────────────────────────────────────────────────┘

Each worker embeds Kafka producer and consumer clients: source tasks use the producer to write records, and sink tasks use the consumer to read them. In distributed mode, the cluster’s state lives in three internal Kafka topics: connector and task configurations in connect-configs, source-connector offsets in connect-offsets, and connector/task status in connect-status. This Kafka-backed state is what enables fault tolerance—if a worker dies, the remaining workers rebalance its tasks and resume from the last committed offset.

Task Rebalancing and Failure Handling

When a worker joins or leaves the group, Connect runs a rebalance protocol built on the Kafka group-membership machinery. By default, since Kafka 2.3 (KIP-415), distributed workers use the incremental cooperative rebalancing protocol, which reassigns only the tasks that must move rather than stopping every task on every membership change. To absorb transient worker restarts without an immediate reassignment, set scheduled.rebalance.max.delay.ms (default 300000, i.e. five minutes): the leader holds a departed worker’s tasks unassigned for up to this delay, giving the worker time to return before its load is redistributed.

Task failure handling is governed by the connector’s errors.tolerance setting. all causes the task to skip records that fail conversion or transformation and continue; none (the default) fails the task on the first such error. Production pipelines often combine errors.tolerance=all with a dead letter queue to capture problematic records for later inspection. The DLQ is configured at the connector level (it applies to sink connectors):

{
  "errors.tolerance": "all",
  "errors.deadletterqueue.topic.name": "my-connector-dlq",
  "errors.deadletterqueue.context.headers.enable": true,
  "errors.deadletterqueue.topic.replication.factor": 3
}

Note that the DLQ captures records that fail conversion, deserialization, or SMT processing—not records rejected by the downstream system inside a connector’s own logic. This pattern is valuable when integrating with legacy systems where malformed data is a known risk—a scenario covered in Developing a Custom Source Connector for Legacy Systems.

Distributed Mode: The Production Standard

Distributed mode is the deployment topology for production. Workers form a group, share configuration state through the internal Kafka topics, and dynamically balance task execution. It provides scalability, fault tolerance, and a single REST API for managing every connector in the cluster.

Cluster Bootstrap and Configuration

A distributed worker starts from a properties file that specifies the Kafka bootstrap servers, the group.id that identifies the cluster, and the internal topics used for coordination. A minimal production worker configuration:

bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
group.id=connect-cluster-prod
key.converter=io.confluent.connect.avro.AvroConverter
value.converter=io.confluent.connect.avro.AvroConverter
key.converter.schema.registry.url=http://schema-registry:8081
value.converter.schema.registry.url=http://schema-registry:8081
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
plugin.path=/usr/share/java,/opt/connectors
rest.advertised.host.name=connect-worker-1

The config.storage.topic must be a single-partition, compacted topic; the offset and status topics should also be compacted. The plugin.path directive tells the worker where to find connector plugins. In containerised environments this is usually a mounted volume or a directory baked into the image. The Confluent Hub client is a common way to install connectors:

confluent-hub install confluentinc/kafka-connect-jdbc:10.7.0 \
  --component-dir /opt/connectors \
  --worker-configs /etc/kafka/connect-distributed.properties

REST API and Operational Commands

A distributed worker exposes a REST API on port 8083 by default. It is the primary interface for deploying, updating, and monitoring connectors:

# List active connectors
curl -s http://connect-worker:8083/connectors | jq .

# Get connector and task status
curl -s http://connect-worker:8083/connectors/my-source-connector/status | jq .

# Create a new connector
curl -X POST http://connect-worker:8083/connectors \
  -H "Content-Type: application/json" \
  -d @my-connector-config.json

# Restart a single failed task
curl -X POST http://connect-worker:8083/connectors/my-source-connector/tasks/0/restart

Any worker can serve any request: the worker that receives a write forwards it to the cluster leader as needed and the change propagates through the config topic. As of Kafka 3.0 (KIP-745), POST /connectors/{name}/restart also accepts ?includeTasks=true&onlyFailed=true to restart a connector and its failed tasks in one call. Because all workers are interchangeable for reads, you can put a reverse proxy in front of the cluster and route /connectors traffic to any of them.

Scaling and Capacity Planning

Horizontal scaling is simple: start additional worker processes with the same group.id and the same internal-topic settings. New workers join the group, and the cluster rebalances tasks across the larger pool with no restart of existing workers.

Scaling is bounded by tasks.max. For a JDBC source reading a table with a monotonically increasing key, you might configure:

{
  "name": "jdbc-source-orders",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "4",
    "mode": "incrementing",
    "incrementing.column.name": "order_id",
    "topic.prefix": "orders-",
    "poll.interval.ms": "5000"
  }
}

tasks.max is an upper bound, not a guarantee: the connector decides how many tasks to create, and many produce fewer. The standard JDBC source, for instance, parallelises across the tables it ingests, so a single-table job like the one above runs only one task regardless of tasks.max. Once a connector’s task count is reached, extra workers stay idle for that connector (though they can still run tasks from others). Confluent’s Connect tuning guide covers worker thread pools and JVM heap sizing for high-throughput pipelines.

Standalone Mode: Development and Simple Pipelines

Standalone mode runs a single worker that executes all of its connectors and tasks locally. Connector configurations are supplied as files on startup rather than through the REST API, and there is no fault tolerance or dynamic rebalancing. It suits development, testing, and edge use cases where high availability is not required.

A standalone worker takes the worker properties file followed by one or more connector configuration files:

connect-standalone /etc/kafka/connect-standalone.properties \
  /etc/kafka/connect-file-source.properties \
  /etc/kafka/connect-file-sink.properties

The worker properties file is simpler than the distributed equivalent because there is no group coordination or internal topics:

bootstrap.servers=kafka1:9092
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=false
value.converter.schemas.enable=false
offset.storage.file.filename=/tmp/connect.offsets

Standalone mode persists source offsets to a local file rather than a Kafka topic, so offset state is tied to that machine and cannot be handed to another worker. This makes standalone a poor fit for durable production pipelines but a good fit for edge deployments and CI where simplicity matters. (Standalone still exposes the REST API on port 8083 by default, but it cannot be used to create or delete connectors.)

Deployment Topologies and Infrastructure Patterns

The choice of topology shapes operability, resource utilisation, and failure domains. Three patterns dominate.

Dedicated Connect Clusters

Each team or data domain operates its own Connect cluster, providing strong isolation: a misbehaving connector cannot starve connectors owned by another team. The cost is operational overhead—every cluster must be monitored, patched, and scaled independently.

Dedicated clusters are most appropriate when connectors have very different resource profiles. A Change Data Capture with Debezium and Kafka pipeline tailing MySQL binlogs at high volume should not share a JVM with a lightweight HTTP sink. Debezium’s memory footprint while snapshotting large tables can drive garbage-collection pressure that degrades the latency of co-located connectors.

Shared Multi-Tenant Clusters

A single large cluster runs connectors for multiple teams. This maximises resource efficiency but demands governance. Connector configs must be vetted to prevent resource exhaustion, and you should choose a deliberate value for the worker property connector.client.config.override.policy, which controls whether a connector can override the worker’s producer, consumer, and admin client settings via the producer.override., consumer.override., and admin.override. prefixes.

This default changed in Kafka 3.0 (KIP-722): it is now All (any override allowed), whereas earlier versions defaulted to None. On a shared cluster, prefer the stricter None to block all overrides, or Principal to allow only security.protocol, sasl.jaas.config, and sasl.mechanism overrides so tenants can supply per-connector credentials without touching performance-sensitive settings.

To enforce richer rules you can supply a custom policy class. The interface, introduced in KIP-458, is ConnectorClientConfigOverridePolicy; its validate method receives the requested client overrides and returns a List<ConfigValue> carrying any errors:

public class StrictOverridePolicy implements ConnectorClientConfigOverridePolicy {
    @Override
    public List<ConfigValue> validate(ConnectorClientConfigRequest request) {
        List<ConfigValue> results = new ArrayList<>();
        for (Map.Entry<String, Object> e : request.clientProps().entrySet()) {
            ConfigValue cv = new ConfigValue(e.getKey());
            cv.value(e.getValue());
            // Reject any attempt to weaken security.protocol
            if ("security.protocol".equals(e.getKey())
                    && !"SASL_SSL".equals(e.getValue())) {
                cv.addErrorMessage("security.protocol override must remain SASL_SSL");
            }
            results.add(cv);
        }
        return results;
    }

    @Override public void configure(Map<String, ?> configs) { }
    @Override public void close() { }
}

Note that this policy governs client overrides only; it cannot see connector-level keys such as tasks.max. Enforcing limits on tasks.max requires a separate validation step (for example, a gateway in front of the REST API or a custom Connector.validate() override), not a ConnectorClientConfigOverridePolicy.

Kubernetes-Native Deployments

Running Connect on Kubernetes is the default for many organisations. The Strimzi operator manages Connect clusters declaratively, handling configuration, scaling, and rolling updates through custom resources. A KafkaConnect resource specifies the cluster, and the build section lets the operator assemble a connector image:

apiVersion: kafka.strimzi.io/v1beta2
kind: KafkaConnect
metadata:
  name: my-connect-cluster
  annotations:
    strimzi.io/use-connector-resources: "true"
spec:
  replicas: 3
  bootstrapServers: kafka-cluster-kafka-bootstrap:9092
  config:
    group.id: connect-cluster
    config.storage.replication.factor: 3
    offset.storage.replication.factor: 3
    status.storage.replication.factor: 3
  build:
    output:
      type: docker
      image: registry.example.com/my-connect:latest
    plugins:
      - name: debezium-mysql
        artifacts:
          - type: tgz
            url: https://repo1.maven.org/maven2/io/debezium/debezium-connector-mysql/2.5.0.Final/debezium-connector-mysql-2.5.0.Final-plugin.tar.gz

The strimzi.io/use-connector-resources annotation enables managing connectors as KafkaConnector custom resources instead of through the REST API. For a full walkthrough, see Deploying Kafka Connect on Kubernetes with Strimzi. Because Connect distributes tasks across whatever workers are present, you can also drive replica count from the Kubernetes Horizontal Pod Autoscaler on CPU or custom metrics—keeping in mind that each scaling event triggers a rebalance.

Converter and Serialization Architecture

Converters bridge the Connect runtime and the Kafka wire format. Every connector resolves a key.converter and value.converter (from the connector config if present, otherwise the worker default). The converter choice has far-reaching consequences for schema management, message size, and interoperability.

Avro and Schema Registry Integration

The Avro converter, used with Schema Registry, is the common choice for production pipelines. It serialises records as Avro binary and stores the schema in Schema Registry, embedding only a schema ID in each message—minimising payload size and enforcing compatibility. The converter is set at the worker level but can be overridden per connector:

{
  "key.converter": "io.confluent.connect.avro.AvroConverter",
  "value.converter": "io.confluent.connect.avro.AvroConverter",
  "key.converter.schema.registry.url": "http://schema-registry:8081",
  "value.converter.schema.registry.url": "http://schema-registry:8081"
}

With the Avro converter, every record a source connector emits must carry a Connect schema; the converter maps that schema to Avro and registers it. Sink connectors do the reverse, fetching the registered schema to deserialise incoming records. This coupling makes Schema Evolution and Compatibility in Confluent Schema Registry a prerequisite topic for any team operating Connect at scale.

JSON and Protobuf Converters

The built-in JsonConverter is simpler but less efficient. With schemas.enable=true it embeds a schema envelope in every message, which inflates payload size; for lightweight integrations where schema evolution is not a concern, schemas.enable=false and plain JSON is acceptable.

The Protobuf converter is a middle ground: compact binary serialisation with schema management via Schema Registry, popular in gRPC-oriented ecosystems. Its class is io.confluent.connect.protobuf.ProtobufConverter, and it requires the Protobuf schema to be resolvable in Schema Registry.

Single Message Transforms

Single Message Transforms (SMTs) are lightweight functions that modify records as they flow through the pipeline. They run in the worker JVM—applied after a source task produces a record and before serialisation, and after deserialisation and before a sink task writes. Common built-ins include InsertField, ReplaceField, and TimestampRouter.

SMTs are configured on the connector using the transforms prefix:

{
  "transforms": "addMetadata,routeTopic",
  "transforms.addMetadata.type": "org.apache.kafka.connect.transforms.InsertField$Value",
  "transforms.addMetadata.offset.field": "_offset",
  "transforms.addMetadata.partition.field": "_partition",
  "transforms.routeTopic.type": "org.apache.kafka.connect.transforms.TimestampRouter",
  "transforms.routeTopic.topic.format": "${topic}-${timestamp}",
  "transforms.routeTopic.timestamp.format": "yyyy-MM-dd"
}

The InsertField$Value.offset.field option populates only on the sink side (it draws from the consumed record’s Kafka offset); on the source side use InsertField$Value for static or per-record metadata instead. SMTs execute on every record, so chains of expensive transforms become a throughput bottleneck. For heavy transformation logic, prefer a stream processing layer such as Kafka Streams or ksqlDB downstream of Connect.

Monitoring and Observability

Operating Connect in production requires visibility into worker health, rebalance activity, and per-task throughput and errors. Connect exposes metrics through JMX, which Prometheus can scrape via the JMX Exporter.

Key Metrics to Track

These metrics are essential for operational monitoring, as documented in the Apache Kafka Connect monitoring reference. The names below are MBean type → attribute:

All of these live under the kafka.connect JMX domain—for example, the full object name for the first row is kafka.connect:type=connect-worker-metrics, and the task-scoped beans add connector="..." and task="..." key properties.

A Prometheus JMX Exporter configuration for Connect:

lowercaseOutputName: true
rules:
  - pattern: 'kafka.connect<type=connect-worker-metrics><>([^:]+)'
    name: kafka_connect_worker_$1
  - pattern: 'kafka.connect<type=task-error-metrics, connector=(.+), task=(.+)><>([^:]+)'
    name: kafka_connect_task_error_$3
    labels:
      connector: "$1"
      task: "$2"

Logging and Diagnostic Endpoints

Connect’s log level can be changed at runtime through the REST API without restarting the worker (KIP-495):

# Raise the sink-task logger to DEBUG
curl -X PUT http://connect-worker:8083/admin/loggers/org.apache.kafka.connect.runtime.WorkerSinkTask \
  -H "Content-Type: application/json" \
  -d '{"level": "DEBUG"}'

GET /admin/loggers lists the current loggers and levels. Two caveats: by default the change applies only to the worker that received the request, not the whole cluster (cluster-wide scope was added later via the ?scope=cluster query parameter in KIP-976), and the change is not persisted across restarts. Setting an ancestor logger such as org.apache.kafka.connect updates all child loggers.

For sink connectors, consumer lag is the key throughput signal. A sink connector’s consumer group ID is connect-{connector-name}, and its lag is visible through the standard tool:

kafka-consumer-groups --bootstrap-server kafka1:9092 \
  --group connect-elasticsearch-sink \
  --describe

Operational Best Practices and Failure Modes

Production Connect clusters fail in predictable ways. Recognising these modes and their mitigations is the mark of a mature operational practice.

Offset Management and Delivery Semantics

Source connectors commit offsets to connect-offsets every offset.flush.interval.ms (default 60000). If a task crashes between commits, it reprocesses records on restart, yielding at-least-once delivery. Exactly-once source support exists as of Kafka 3.3 (KIP-618) and is enabled with exactly.once.source.support=enabled on the worker, provided the connector implements it; otherwise, make the destination idempotent.

The JDBC sink, for example, achieves idempotency through upsert semantics:

{
  "insert.mode": "upsert",
  "pk.mode": "record_key",
  "delete.enabled": false
}

Connector Plugins and Class Loading

Connect loads each plugin in an isolated classloader to prevent dependency conflicts. The isolation is not absolute: if two plugins bundle incompatible versions of a transitively shared library on the same path, NoClassDefFoundError or LinkageError can still surface. Give each plugin its own directory under plugin.path, and prefer plugins that shade their dependencies.

When deploying database connectors, such as those in Streaming MySQL Changes to Kafka with Debezium, place the JDBC driver JAR inside that connector’s plugin directory rather than on the worker’s main classpath, so it loads in the plugin’s isolated classloader and cannot collide with another connector’s driver version.

Graceful Shutdown and Rolling Restarts

A worker that receives SIGTERM shuts down gracefully: it stops taking new assignments, finishes in-flight records, commits offsets, and leaves the group. The shutdown budget is task.shutdown.graceful.timeout.ms (default 5000)—and note this is the total budget for all of a worker’s tasks, which are stopped sequentially, not a per-task value. Increase it for connectors that process large batches so shutdown doesn’t time out before offsets are committed.

For rolling restarts, restart one worker at a time and wait for the cluster to stabilise before the next. “Stable” means the rebalance has settled: connect-worker-rebalance-metrics → rebalance-avg-time-ms is no longer moving, the connector-count/task-count totals across surviving workers match expectations, and GET /connectors/{name}/status shows all tasks RUNNING. With incremental cooperative rebalancing and a non-zero scheduled.rebalance.max.delay.ms, a brief restart may not trigger a full reassignment at all.

Security Configuration

Production Connect should authenticate to Kafka with SASL and encrypt traffic with TLS. The worker carries the standard Kafka client security properties:

security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="connect" \
  password="connect-secret";
ssl.truststore.location=/etc/security/connect.truststore.jks
ssl.truststore.password=truststore-password

Connectors that reach external systems should pull credentials from a secrets manager rather than embedding them in the connector config. The ConfigProvider interface (KIP-297) resolves externalised values at runtime. Apache Kafka ships a built-in FileConfigProvider; for HashiCorp Vault, the widely used community provider’s class is com.github.jcustenborder.kafka.config.vault.VaultConfigProvider (verify the exact class against the build you install, as Confluent’s CSID provider uses a different package):

{
  "config.providers": "vault",
  "config.providers.vault.class": "com.github.jcustenborder.kafka.config.vault.VaultConfigProvider",
  "config.providers.vault.param.vault.address": "https://vault.internal:8200",
  "db.password": "${vault:secret/database:password}"
}

Resolution happens at task startup, so the literal secret never appears in plaintext in the connect-configs topic—only the ${vault:...} reference is stored.

Backup and Disaster Recovery

A distributed cluster’s entire state lives in connect-configs, connect-offsets, and connect-status. Backing up a Connect cluster therefore reduces to protecting those topics (and, ideally, keeping connector configs in version control). In a recovery scenario, a new cluster pointed at replicas of these topics resumes from where the old one left off—provided source/sink offsets are valid for the destination Kafka cluster.

Treat connector configurations as infrastructure as code: store them in version control and apply them through a pipeline that calls the REST API. This makes the cluster reproducible and every configuration change auditable.