Distributed Worker Setup

Summary

Distributed mode is more complex to configure but provides automatic task rebalancing, shared configuration storage, and horizontal scalability. Multiple workers form a cluster and coordinate using Kafka topics.

In distributed mode, multiple workers coordinate to run connectors and tasks. Configuration is shared through Kafka topics, not local files.

All workers connect to the same Kafka cluster and use internal topics to synchronize state. You can send REST API requests to any worker, and they'll all see the same configuration.

Worker Configuration

Let's configure connect-distributed.properties. This file defines how workers coordinate:

1# Kafka broker connection
2bootstrap.servers=kafka1:9092,kafka2:9092,kafka3:9092
3
4# Unique group ID for this Connect cluster
5group.id=connect-cluster
6
7# Topics for storing connector state
8config.storage.topic=connect-configs
9offset.storage.topic=connect-offsets
10status.storage.topic=connect-status
11
12# Replication factor for internal topics (production: 3)
13config.storage.replication.factor=3
14offset.storage.replication.factor=3
15status.storage.topic.replication.factor=3
16
17# Converters
18key.converter=org.apache.kafka.connect.json.JsonConverter
19value.converter=org.apache.kafka.connect.json.JsonConverter
20key.converter.schemas.enable=false
21value.converter.schemas.enable=false
22
23# REST API configuration
24rest.port=8083
25rest.advertised.host.name=worker1.example.com
26rest.advertised.port=8083
27
28# Plugin path
29plugin.path=/opt/kafka/plugins

The group.id is critical—all workers in the same cluster must use the same group.id. The internal topics store configuration, offsets, and status. High replication factors ensure fault tolerance.

Internal Topics Configuration

The three internal topics have specific purposes:

connect-configs

• Stores connector and task configurations
• Compacted topic (retains latest config for each connector)
• Replication factor should be 3 in production

connect-offsets

• Stores source connector offsets (position in source system)
• Compacted topic (retains latest offset for each partition)
• Very important for exactly-once or at-least-once semantics

connect-status

• Stores connector and task status
• Compacted topic
• Used for health monitoring and coordination

Creating Internal Topics Manually

While Connect can auto-create these topics, it's better to create them manually in production for control:

1# Create configs topic
2bin/kafka-topics.sh --bootstrap-server localhost:9092 \
3  --create \
4  --topic connect-configs \
5  --replication-factor 3 \
6  --partitions 1 \
7  --config cleanup.policy=compact
8
9# Create offsets topic
10bin/kafka-topics.sh --bootstrap-server localhost:9092 \
11  --create \
12  --topic connect-offsets \
13  --replication-factor 3 \
14  --partitions 25 \
15  --config cleanup.policy=compact
16
17# Create status topic
18bin/kafka-topics.sh --bootstrap-server localhost:9092 \
19  --create \
20  --topic connect-status \
21  --replication-factor 3 \
22  --partitions 5 \
23  --config cleanup.policy=compact

Notice the partition counts: configs only needs 1 partition since there aren't many connectors, but offsets uses 25 partitions to handle high throughput from many source connectors.

Starting the First Worker

Start the first worker in the cluster:

1bin/connect-distributed.sh config/connect-distributed.properties

Unlike standalone mode, you don't provide connector configs on the command line. Instead, you'll deploy connectors via REST API after the worker starts.

Starting Additional Workers

On other machines, use the same configuration file with different REST advertised host names:

On worker2:

1# In connect-distributed.properties, set:
2rest.advertised.host.name=worker2.example.com
3
4bin/connect-distributed.sh config/connect-distributed.properties

On worker3:

1# In connect-distributed.properties, set:
2rest.advertised.host.name=worker3.example.com
3
4bin/connect-distributed.sh config/connect-distributed.properties

All workers use the same group.id and internal topics, so they automatically discover each other and form a cluster. No manual coordination needed.

Here's what happens when workers start and join the cluster:

Workers use Kafka's consumer group protocol to coordinate. One worker becomes the leader and handles task assignment. When workers join or leave, a rebalance automatically redistributes tasks.

Verifying the Cluster

Check that all workers joined the cluster by querying the REST API:

1# On any worker
2curl http://localhost:8083/
3
4# Response shows worker version and git commit
5{
6  "version": "3.6.0",
7  "commit": "...",
8  "kafka_cluster_id": "..."
9}

Deploying a Connector

deploy a connector via REST API:

1curl -X POST http://localhost:8083/connectors \
2  -H "Content-Type: application/json" \
3  -d '{
4    "name": "file-source-distributed",
5    "config": {
6      "connector.class": "FileStreamSource",
7      "tasks.max": "3",
8      "file": "/tmp/test.txt",
9      "topic": "connect-test-distributed"
10    }
11  }'

This POST request creates the connector. The worker writes the configuration to the connect-configs topic, all workers see it, and tasks are distributed across the cluster.

Checking Connector Status

Verify the connector is running on any worker:

1curl http://localhost:8083/connectors/file-source-distributed/status

Response:

1{
2  "name": "file-source-distributed",
3  "connector": {
4    "state": "RUNNING",
5    "worker_id": "worker1.example.com:8083"
6  },
7  "tasks": [
8    {
9      "id": 0,
10      "state": "RUNNING",
11      "worker_id": "worker1.example.com:8083"
12    }
13  ]
14}

Production Best Practices

For production distributed deployments:

Resource Allocation

• Minimum 3 workers for fault tolerance
• 4GB+ heap per worker
• Separate machines or containers
• Different availability zones if possible

Network Configuration

• Ensure workers can reach all Kafka brokers
• Open REST API port (8083) for management
• Consider load balancer for REST API

Monitoring

• Monitor internal topic lag
• Track connector and task status
• Alert on failed tasks
• Monitor JVM metrics

Security

• Enable authentication for REST API
• Use SSL for Kafka connections
• Secure internal topics with ACLs
• Protect connector configurations (secrets)

Distributed mode provides the fault tolerance and scalability needed for production. Multiple workers share the workload, and if one fails, the others automatically take over.