3.2 – Standalone vs Distributed Mode

Overview

This module is the authoritative explanation of Kafka Connect's deployment modes. By the end, you will understand:

• Standalone vs. Distributed mode architecture.
• When to use each mode.
• Worker groups and task coordination in distributed mode.
• Internal Kafka topics used for distributed mode.

No other module re-explains deployment modes. Subsequent modules will reference this when discussing scaling, fault tolerance, or operations.

Kafka Connect Deployment Modes

Kafka Connect supports two deployment modes:

1. Standalone Mode: Single-worker, single-process deployment.
2. Distributed Mode: Multi-worker, clustered deployment with fault tolerance.

Key Decision: Choose Standalone for development/testing and simple use cases. Choose Distributed for production deployments requiring scalability and high availability.

Standalone Mode

Architecture

┌───────────────────────────────┐
│     Kafka Connect Worker     │
│                               │
│  ┌─────────┐  ┌─────────┐   │
│  │Connector│  │Connector│   │
│  │  Task 1 │  │  Task 1 │   │
│  │  Task 2 │  │  Task 2 │   │
│  └─────────┘  └─────────┘   │
│                               │
│  Offset Storage: Local File  │
└───────────────────────────────┘

Characteristics

• Single JVM Process: All connectors and tasks run in one worker.
• No Fault Tolerance: If the worker crashes, all connectors stop until manually restarted.
• No Coordination: No worker-to-worker communication.
• Local Offset Storage: Offsets are stored in a local file (not in Kafka).
• No REST API (by default): Configuration is via properties files, not REST.

Configuration

Worker Config (connect-standalone.properties):

# Kafka cluster connection
bootstrap.servers=localhost:9092

# Converters (default for all connectors)
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

# Offset storage (local file)
offset.storage.file.filename=/tmp/connect.offsets

# Plugin path (where connector JARs are located)
plugin.path=/usr/local/share/kafka/plugins

Connector Config (jdbc-source.properties):

name=jdbc-source-standalone
connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
tasks.max=1
connection.url=jdbc:postgresql://localhost:5432/mydb
mode=incrementing
incrementing.column.name=id
topic.prefix=postgres-

Starting Standalone Mode

bin/connect-standalone.sh config/connect-standalone.properties \
  config/jdbc-source.properties \
  config/elasticsearch-sink.properties

Command Breakdown:

• First argument: Worker configuration.
• Subsequent arguments: Connector configurations (one or more).

When to Use Standalone Mode

✅ Use Standalone Mode For:

• Development/Testing: Quick local testing of connectors.
• Single-Connector Deployments: Simple ETL jobs that don't require fault tolerance.
• Low-Volume Workloads: Connectors processing small amounts of data.

❌ Avoid Standalone Mode For:

• Production Deployments: No fault tolerance or scalability.
• High Availability: Single point of failure.
• Dynamic Reconfiguration: Requires process restart to update connectors.

Distributed Mode

Architecture

┌─────────────────────────────────────────────────────────┐
│             Kafka Connect Cluster                       │
│                                                           │
│  ┌────────────┐  ┌────────────┐  ┌────────────┐        │
│  │  Worker 1  │  │  Worker 2  │  │  Worker 3  │        │
│  │            │  │            │  │            │        │
│  │  Task 1.1  │  │  Task 1.2  │  │  Task 2.1  │        │
│  │  Task 2.2  │  │            │  │            │        │
│  └────────────┘  └────────────┘  └────────────┘        │
│         │                 │                 │            │
└─────────┼─────────────────┼─────────────────┼──────────┘
          │                 │                 │
          ▼                 ▼                 ▼
    ┌─────────────────────────────────────────────┐
    │         Internal Kafka Topics               │
    │  - connect-configs                          │
    │  - connect-offsets                          │
    │  - connect-status                           │
    └─────────────────────────────────────────────┘

Characteristics

• Multi-Worker Cluster: Multiple workers coordinate to distribute tasks.
• Fault Tolerance: If a worker fails, its tasks are reassigned to other workers.
• Scalability: Add more workers to handle more connectors/tasks.
• Shared State via Kafka: Connector configs, offsets, and status are stored in internal Kafka topics.
• REST API: Configure and manage connectors via HTTP (any worker can handle requests).

Configuration

Worker Config (connect-distributed.properties):

# Kafka cluster connection
bootstrap.servers=localhost:9092

# Worker group ID (all workers in a cluster must have the same group.id)
group.id=connect-cluster

# Internal topics for state storage
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status

# Replication factors for internal topics (should be ≥ 3 in production)
config.storage.replication.factor=3
offset.storage.replication.factor=3
status.storage.replication.factor=3

# Converters (default for all connectors)
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
key.converter.schemas.enable=true
value.converter.schemas.enable=true

# Plugin path
plugin.path=/usr/local/share/kafka/plugins

# REST API port (default: 8083)
rest.port=8083

Starting Distributed Mode

On each worker node:

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

Note: All workers with the same group.id form a single Connect cluster.

Environment Variables vs Properties Files

Kafka Connect supports two approaches for configuring distributed mode workers:

Approach 1: Properties File

Create connect-distributed.properties:

bootstrap.servers=kafka:9092
rest.port=8083
group.id=connect-cluster
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
key.converter=org.apache.kafka.connect.json.JsonConverter
value.converter=org.apache.kafka.connect.json.JsonConverter
plugin.path=/usr/local/share/kafka/plugins

Start Worker:

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

Approach 2: Environment Variables (Docker/Kubernetes Best Practice)

In Docker or Kubernetes deployments, use environment variables instead of properties files:

Docker Compose Example:

kafka-connect:
  image: confluentinc/cp-kafka-connect:7.5.0
  environment:
    CONNECT_BOOTSTRAP_SERVERS: kafka:9092
    CONNECT_REST_PORT: 8083
    CONNECT_GROUP_ID: connect-cluster
    CONNECT_CONFIG_STORAGE_TOPIC: connect-configs
    CONNECT_OFFSET_STORAGE_TOPIC: connect-offsets
    CONNECT_STATUS_STORAGE_TOPIC: connect-status
    CONNECT_KEY_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    CONNECT_VALUE_CONVERTER: org.apache.kafka.connect.json.JsonConverter
    CONNECT_PLUGIN_PATH: /usr/share/java

Mapping Rule: property.name → CONNECT_PROPERTY_NAME

Examples:

• bootstrap.servers → CONNECT_BOOTSTRAP_SERVERS
• group.id → CONNECT_GROUP_ID
• rest.port → CONNECT_REST_PORT
• config.storage.topic → CONNECT_CONFIG_STORAGE_TOPIC

Why Use Environment Variables:

• Easier containerization: No need to mount properties files.
• Dynamic configuration: Change settings without rebuilding images.
• Cloud-native: Standard practice in Kubernetes, ECS, Cloud Run, etc.
• Secret management: Integrate with Docker secrets, Kubernetes secrets, etc.

Best Practice: Use environment variables for production deployments in containers, properties files for traditional VM/bare-metal deployments.

When to Use Distributed Mode

✅ Use Distributed Mode For:

• Production Deployments: Fault tolerance and high availability.
• Multi-Connector Workloads: Scale across multiple workers.
• Dynamic Reconfiguration: Add/update/delete connectors via REST API without restarts.
• High-Volume Data Pipelines: Distribute tasks across workers for higher throughput.

❌ Avoid Distributed Mode For:

• Local Development: Overhead of managing internal Kafka topics and multiple workers.

Worker Groups

What is a Worker Group?

A worker group is a set of Connect workers that share the same group.id and coordinate to distribute connector tasks.

How Worker Groups Work

1. Group Membership: Workers with the same group.id join the same group.
2. Leader Election: One worker is elected as the group leader (responsible for task assignment).
3. Task Distribution: The leader assigns tasks to workers based on capacity and availability.
4. Rebalancing: If a worker joins or leaves, tasks are reassigned to maintain balance.

Analogy: Worker groups are like server pools in a load balancer—each server (worker) can handle multiple requests (tasks), and the load balancer (group leader) distributes work evenly.

Example: 3-Worker Cluster

Setup:

• group.id=connect-cluster
• 3 Workers: worker-1, worker-2, worker-3
• Connector: jdbc-source with tasks.max=6

Task Assignment:

Rebalance Scenario: If worker-2 crashes, its tasks are reassigned:

Internal Kafka Topics (Distributed Mode Only)

Distributed mode uses three internal Kafka topics to store shared state. Understanding these topics is critical for operating Kafka Connect in production.

1. connect-configs

• Purpose: Stores connector and task configurations.
• Format: Compacted topic (log compaction retains only the latest value per key).
• Key: Connector name or task ID.
• Value: JSON configuration.

Example Record:

Key: "jdbc-source"
Value: {
  "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
  "tasks.max": "3",
  "connection.url": "jdbc:postgresql://localhost:5432/mydb"
}

Why This Matters:

Kafka Connect internally stores all connector configs inside a Kafka topic to ensure they survive restarts and are shared across all workers.

Key Benefits:

1. Persistence: You don't lose connector configs even if Connect crashes—they are safely saved in the connect-configs topic.
2. Shared State: All workers in the cluster read from this topic, ensuring consistent configuration.
3. Audit Trail: The topic serves as a log of all connector configurations over time.

What Happens on Restart:

1. Worker starts up.
2. Worker consumes entire connect-configs topic from beginning.
3. Worker rebuilds its in-memory state of all connectors.
4. Worker is ready to accept new configs or execute tasks.

2. connect-offsets

• Purpose: Stores source connector offsets (equivalent to standalone's local offset file).
• Format: Compacted topic.
• Key: Connector name + partition (logical partition within the source).
• Value: Offset data (e.g., database row ID, file position).

Example Record:

Key: ["jdbc-source", {"table": "users"}]
Value: {"id": 12345}

Why This Matters:

Kafka Connect needs to track how much data it has read from sources (or written to sinks). The offsets ("I've already read until line 1000 of this file" or "I've processed database row with id=12345") are saved into a Kafka topic.

Key Benefits:

1. Resumption: When an instance of Kafka Connect restarts, it resumes from where it left off, thanks to these saved offsets.
2. No Data Loss: Offsets are durably stored in Kafka, not in local files.
3. Exactly-Once Semantics: Enables exactly-once delivery patterns (when combined with idempotent external systems).

What Happens on Restart:

1. Source task starts up.
2. Task reads its last committed offset from connect-offsets.
3. Task resumes polling from that offset (e.g., SELECT * FROM users WHERE id > 12345).
4. New offsets are committed as data is processed.

Contrast with Standalone Mode: In standalone mode, offsets are stored in a local file (e.g., /tmp/connect.offsets), which is lost if the file is deleted or the server fails.

3. connect-status

• Purpose: Stores connector and task status (running, failed, paused).
• Format: Compacted topic.
• Key: Connector name or task ID.
• Value: Status metadata (state, worker ID, error messages).

Example Record:

Key: "jdbc-source"
Value: {
  "state": "RUNNING",
  "worker_id": "worker-1:8083",
  "generation": 5
}

Why This Matters:

Kafka Connect needs to keep track of the status of each connector and task. This lets you see if a connector is running, failed, or paused. You can check health and manage connectors easily through the REST API.

Key Benefits:

1. Health Monitoring: Query connector/task status via REST API without querying each worker directly.
2. Coordination: Workers coordinate task assignments and failover via this topic.
3. Observability: Track connector state changes over time (audit trail).

What Happens on Status Change:

1. Connector state changes (e.g., RUNNING → FAILED).
2. Worker writes new status record to connect-status.
3. All workers consume the update.
4. REST API queries reflect the new status immediately.

REST API Integration:

curl http://localhost:8083/connectors/jdbc-source/status

This query reads from the worker's in-memory state, which is built by consuming connect-status.

Understanding group.id

The group.id configuration sets the consumer group ID used by the Kafka Connect cluster.

group.id=connect-cluster

Purpose: When Connect nodes (workers) consume from Kafka topics (like configs, offsets, statuses), they act as a Kafka consumer group.

How It Works:

1. All workers with the same group.id join the same consumer group.
2. They coordinate to share the workload of consuming internal topics.
3. This ensures exactly-once processing of configuration changes.

Example: If you have 3 workers:

• Worker 1: group.id=connect-cluster
• Worker 2: group.id=connect-cluster
• Worker 3: group.id=connect-cluster

All three workers are part of the same Connect cluster and will coordinate task distribution.

Important: Do not use the same group.id for different Connect clusters, or they will interfere with each other!

Contrast with Standalone Mode

In standalone mode, there is only one worker running, so:

• No group.id needed (no consumer group coordination).
• No storage topics needed (configs, offsets, statuses are stored locally).
• Local file storage:
  • offset.storage.file.filename=/tmp/connect.offsets
  • status.storage.file.filename=/tmp/connect.status
  • config.storage.file.filename=/tmp/connect.config

Trade-off: Simplicity vs. fault tolerance. Standalone is simple but fragile; distributed is complex but resilient.

Topic Configuration Best Practices

Note: High replication factor (3+) ensures internal topics survive broker failures.

Partition Count Guidelines:

• connect-configs: 1 partition (order matters for config changes).
• connect-offsets: 25 partitions (default, allows parallelism across connectors).
• connect-status: 5 partitions (balances parallelism and overhead).

Standalone vs Distributed: Detailed Comparison

Task Coordination in Distributed Mode

Rebalancing

Rebalancing is the process of redistributing tasks when:

1. A new worker joins the cluster.
2. An existing worker leaves (graceful shutdown or crash).
3. A connector is reconfigured with a different tasks.max.

Rebalance Protocol:

1. Detection: Workers detect membership changes via Kafka's group coordination protocol.
2. Leader Election: A new group leader is elected (if the old leader left).
3. Task Assignment: The leader computes a new task assignment and writes it to connect-status.
4. Task Start/Stop: Workers start/stop tasks according to the new assignment.

Impact: During rebalancing, affected tasks are briefly paused (typically <10 seconds in a healthy cluster).

Leader Election

Leader Responsibilities:

• Assign connectors and tasks to workers.
• Trigger rebalancing when membership changes.
• Monitor task health and reassign failed tasks.

Follower Responsibilities:

• Execute assigned tasks.
• Report status to the leader via connect-status.

Note: Any worker can become the leader; the role is not permanently assigned.

Choosing the Right Mode

Decision Matrix

Migration Path

Start: Standalone for local testing.
Migrate: To distributed mode for production.

Migration Steps:

1. Set up distributed mode workers (3+ for fault tolerance).
2. Create internal topics (connect-configs, connect-offsets, connect-status).
3. Submit connector configs via REST API.
4. Decommission standalone workers.

Common Operational Tasks

Adding a Worker to a Distributed Cluster

1. Install Kafka Connect on the new node.
2. Configure group.id to match existing cluster.
3. Start the worker: bin/connect-distributed.sh config/connect-distributed.properties
4. Observe automatic rebalancing in logs.

Removing a Worker

Graceful Shutdown:

kill -SIGTERM <worker-pid>

Result: Worker stops tasks, triggers rebalancing, and exits cleanly.

Ungraceful Shutdown (crash):

Result: Other workers detect the failure via heartbeat timeout, trigger rebalancing, and reassign tasks.

Monitoring Worker Health

Check REST API:

curl http://localhost:8083/

Response (healthy):

{
  "version": "3.5.0",
  "commit": "...",
  "kafka_cluster_id": "..."
}

Key Takeaways

1. Standalone mode: Single worker, no fault tolerance, file-based offsets. Use for dev/testing.
2. Distributed mode: Multi-worker cluster, fault-tolerant, Kafka-based state storage. Use for production.
3. Worker groups: Workers with the same group.id coordinate via leader election and rebalancing.
4. Internal topics: connect-configs, connect-offsets, connect-status store shared state in distributed mode.
5. Rebalancing: Automatic task reassignment when workers join/leave or connectors are reconfigured.
6. Best practice: Use distributed mode with ≥3 workers and replication factor ≥3 for internal topics.

What's Next?

• Module 3.3 explains task parallelism and scaling strategies in detail.
• Module 8.1 covers monitoring worker and task health.
• Module 5.1 discusses performance tuning for distributed clusters.

Further Reading

• Kafka Connect Architecture (Confluent)
• Distributed Mode Configuration

End of Module 3.2