Standalone vs Distributed Mode #

Apache Kafka Connect is designed to serve various scales of data movement needs, from simple local testing to processing millions of events per second in enterprise-scale production environments. To meet those needs, Kafka Connect provides two different operational runtime modes: Standalone Mode and Distributed Mode. Choosing the runtime mode determines how inter-process coordination runs, how connector configuration is defined, and how fault tolerance and horizontal scalability are managed. This article will deeply dissect the architectural differences between Standalone and Distributed Mode, unpack the important role of Kafka Connect’s three internal topics in maintaining cluster consistency, and discuss the dynamic rebalancing mechanism keeping data pipelines running without obstacles.

Standalone Mode: Characteristics and Use Cases #

Standalone Mode is the simplest runtime model of Kafka Connect. In this mode, all Connect framework execution—including connector definitions, task division, work coordination, and data serialization—runs inside one single JVM (Java Virtual Machine) process on one physical or virtual server machine.

How It Works and Configuration #

In Standalone Mode, we don’t interact with Kafka Connect using the REST API to create or modify connectors. Instead, all configuration is defined using local text properties files (.properties) read when the worker process first starts.

The terminal command to run a standalone worker typically looks like this:

# Running a standalone worker with one worker config file and one or more connector config files
connect-standalone.sh connect-standalone.properties mysql-source.properties elasticsearch-sink.properties

Inside the connect-standalone.properties file, we define basic worker configuration like Kafka broker connection details and the Converters used. While the other properties files contain connector-specific parameters.

Standalone Mode Advantages #

• Simple and Fast: Very easy to set up because it doesn’t require creating special internal topics in Kafka.
• Lightweight: Uses relatively small JVM heap memory because there’s no inter-machine coordination overhead.
• Perfect for Local Testing: Very ideal for developer teams debugging custom connectors locally on their computers.

Weaknesses and Limitations #

• No Fault Tolerance (No High Availability): If the JVM process crashes (for example from OutOfMemory or machine death), the entire data pipeline immediately stops. Failover must be done manually by restarting the service.
• Limited Scalability: Parallelism is only bounded by the compute power (CPU/Memory) of one single machine. We can’t add new machines to automatically divide workloads.
• Static Configuration Management: Every time we want to add, remove, or change connector configuration, we must shut down the worker, edit the properties file, and restart it (downtime).

Appropriate Usage Scenarios #

Although not suitable for main production environments, Standalone Mode is very reliable for:

• Streaming local log files from edge nodes to Kafka, similar to log collector agents (like Filebeat or Fluentbit).
• One-time data migration processes where high availability isn’t a critical requirement.

Distributed Mode: Multi-Worker Architecture Design #

To meet modern production system needs demanding zero-downtime, elastic scalability, and automatic fault tolerance, Kafka Connect provides Distributed Mode. In this mode, several Connect workers (independent JVM processes) run on different servers forming one unified logical cluster.

flowchart TD
    subgraph ClusterGroup["Kafka Connect Distributed Cluster (group.id = connect-prod)"]
        WorkerA["Worker Node A (REST API: 8083)"]
        WorkerB["Worker Node B (REST API: 8083)"]
        WorkerC["Worker Node C (REST API: 8083)"]
    end
    
    subgraph InternalTopics["Kafka Broker (Shared State Storage)"]
        ConfigTopic[("connect-configs (1 Partition, Compacted)")]
        OffsetTopic[("connect-offsets (Compacted)")]
        StatusTopic[("connect-status (Compacted)")]
    end
    
    Dev["Developer / Admin API Client"] -->|REST Request: Create Connector| WorkerA
    WorkerA -->|Write New Configuration| ConfigTopic
    ConfigTopic -.->|Detect Changes via Consumer| WorkerB & WorkerC
    
    WorkerB -->|Synchronize Write/Read Offsets| OffsetTopic
    WorkerC -->|Publish Health Status| StatusTopic

1. Masterless Coordination (Masterless Architecture) #

The Distributed Mode cluster doesn’t rely on traditional Master-Worker architecture with a single point of failure. Instead, inter-worker coordination is based on Apache Kafka’s internal Group Coordinator, using the same consumer group protocol membership mechanism as regular Kafka consumers.

Every worker started with the same group.id configuration automatically joins the same Connect cluster. One worker is dynamically designated by Kafka as the group Leader, responsible for evenly dividing tasks among all active workers. If that leader worker dies, Kafka instantly designates another worker to take over the leadership role.

2. Dynamic Management via REST API #

In Distributed Mode, we no longer include local connector configuration files when starting workers. We simply start empty workers:

# Starting a distributed worker with only the worker properties file
connect-distributed.sh connect-distributed.properties

After the worker cluster starts, all connector add, modify, monitor, and delete operations happen dynamically through the HTTP REST API protocol exposed by each worker (by default on port 8083).

For example, to create a new MySQL Source Connector, we just send an HTTP POST request with a JSON payload:

curl -X POST -H "Content-Type: application/json" \
     --data '{
       "name": "mysql-source-dynamic",
       "config": {
         "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
         "tasks.max": "3",
         "connection.url": "jdbc:mysql://mysql-db:3306/shop",
         "connection.user": "read_user",
         "connection.password": "pass",
         "mode": "incrementing",
         "incrementing.column.name": "id",
         "topic.prefix": "shop-orders-"
       }
     }' http://localhost:8083/connectors

This REST API makes it easy for DevOps teams to integrate connector deployment into automated CI/CD pipelines without manual intervention on worker operating systems.

REST API Endpoint Reference for Cluster Operations #

Distributed Mode exposes a complete REST API endpoint set on every worker. API clients (like curl, Postman, or automation scripts) can contact any worker IP in the cluster because configuration is synchronized to all nodes in real-time.

Here’s the list of REST API endpoints we must know for operating a Connect cluster:

• GET /connectors

  • Purpose: Displays the list of all active connector names in the cluster.
  • Example Response: ["mysql-source-dynamic", "elasticsearch-sink-logs"]

• POST /connectors

  • Purpose: Creates a new connector. The payload is a JSON object containing the connector name and its configuration properties.

• GET /connectors/{name}

  • Purpose: Fetches the active configuration details of a specific connector.

• GET /connectors/{name}/status

  • Purpose: Checks connector health and its tasks’ status in real-time. This is the most important endpoint for system monitoring.
  • Example Response:

    {
      "name": "mysql-source-dynamic",
      "connector": {
        "state": "RUNNING",
        "worker_id": "192.168.1.50:8083"
      },
      "tasks": [
        {
          "id": 0,
          "state": "RUNNING",
          "worker_id": "192.168.1.50:8083"
        },
        {
          "id": 1,
          "state": "FAILED",
          "trace": "java.sql.SQLException: Connection pool exhausted...",
          "worker_id": "192.168.1.51:8083"
        }
      ],
      "type": "source"
    }
    

• PUT /connectors/{name}/config

  • Purpose: Dynamically changes an existing connector’s configuration. This configuration change automatically triggers a task rebalance to apply the new settings.

• POST /connectors/{name}/restart

  • Purpose: Restarts the Connector instance (for example after configuration was manually changed externally).

• POST /connectors/{name}/tasks/{task_id}/restart

  • Purpose: Restarts a specific FAILED task from transient disruptions without stopping other normally running tasks.

• DELETE /connectors/{name}

  • Purpose: Permanently deletes a connector and frees all tasks running under it.

Distributed Worker Configuration Parameter Dissection #

To optimally configure a Distributed Worker in production, we need to dive into the worker properties file (connect-distributed.properties). Below is the essential parameter configuration with its functional explanation:

# ✓ CORRECT: Configuring bootstrap servers to our production Kafka broker cluster
bootstrap.servers=kafka-broker-1:9092,kafka-broker-2:9092,kafka-broker-3:9092

# Connect cluster group identity. All workers with the same group.id form 1 cluster
group.id=connect-cluster-production

# HTTP REST API port opened for administration queries
listeners=HTTP://0.0.0.0:8083
# Host address promoted to other workers so internal REST coordination runs normally
rest.advertised.host.name=connect-worker-1
rest.advertised.port=8083

# --- INTERNAL TOPIC SETTINGS (Must Have High Replication) ---

# Topic for connector configuration. Must be set to 1 partition!
config.storage.topic=connect-configs-topic
config.storage.replication.factor=3

# Topic for Source Connector offsets. Set quite a few partitions (e.g., 25)
offset.storage.topic=connect-offsets-topic
offset.storage.replication.factor=3
offset.storage.partitions=25

# Topic for task health status.
status.storage.topic=connect-status-topic
status.storage.replication.factor=3
status.storage.partitions=5

The Role of Kafka Connect’s Three Internal Topics #

One of the most frequently asked architectural questions is: How does the Distributed Mode cluster store its state and configuration so it stays consistent among all workers without using external databases or Zookeeper?

The answer is: Kafka Connect stores all cluster state inside internal Apache Kafka topics. When the worker cluster first starts, workers automatically create the following three internal topics (if they don’t already exist) with special retention configuration:

1. The connect-configs Topic #

This topic stores the complete configuration of all connectors registered to the Connect cluster.

• Mandatory Configuration: Must be set with only 1 partition (num.partitions=1) and a compact cleanup policy (cleanup.policy=compact).
• Why exactly 1 partition?: This is absolutely required to guarantee linearity of configuration ordering (total ordering of configuration changes). All Connect workers read this topic from start to finish to reconstruct the exact same connector list in their local memory.

2. The connect-offsets Topic #

This topic is used by all Source Connectors running in the cluster to store the last offset coordinates of external source systems.

• Mandatory Configuration: Set with more partitions (for example default 25 or 50 partitions) for write scalability, using the cleanup.policy=compact policy.
• How It Works: Every time a Source Task successfully sends a data batch, it writes the source offset coordinates to this topic. If the task is moved to another worker from rebalancing, that task can continue its work from the last offset recorded in this topic.

3. The connect-status Topic #

This topic records the latest health status of all running connectors and tasks (whether they’re RUNNING, FAILED, PAUSED, etc.).

• Mandatory Configuration: Many partitions (for example default 5 or 10 partitions) with the cleanup.policy=compact policy.
• How It Works: Every time a task status changes (for example crashing from an error), the latest status is sent to this topic. The results of the /connectors/{name}/status REST API query are taken directly from the compacted data in this topic.

[!WARNING] In production environments, these three internal topics must be created with a minimum replication factor of 3 (replication.factor=3) and a minimum synchronized replica count of 2 (min.insync.replicas=2). Data loss on the connect-configs or connect-offsets topics destroys the entire integration pipeline state and can potentially cause massive duplicate data leaks.

Rebalancing and Fault Tolerance Mechanisms #

One of Distributed Mode’s most superior features is dynamic fault tolerance through a mechanism called Rebalancing.

How Does Rebalance Happen? #

When a Connect cluster membership change happens—for example, a new worker joins, an old worker crashes (marked by lost heartbeats / heartbeat timeout), or a new connector configuration is registered—the cluster triggers a rebalancing process.

During rebalancing:

1. The leader worker safely stops the old task distribution.
2. The leader recalculates the optimal distribution based on the active worker count and the tasks.max parallelism limit set in the connector configuration.
3. Tasks are redistributed evenly to all available workers.

Incremental Cooperative Rebalancing #

In early Kafka Connect versions (before version 2.3), rebalancing used the blocking Eager Rebalance protocol (stop-the-world rebalance). During the rebalancing process, all tasks in the Connect cluster were shut down first before being reassigned. This caused disruptive latency spikes on time-sensitive data pipelines.

Starting from version 2.3 and above, Kafka Connect switched to the Incremental Cooperative Rebalancing protocol. This new protocol is very smart:

• Only tasks that need moving (because the worker they run on died) are stopped.
• Other tasks running on healthy workers keep running normally without interruption.
• This dramatically reduces cluster downtime and minimizes message duplication from reprocessing old offsets.

Standalone vs Distributed Comparison Table #

Summary #

• Standalone Mode — The best choice for local development, debugging, and simple single-machine integration scenarios, using static local file-based configuration.
• Distributed Mode — The mandatory production solution offering automatic fault tolerance (failover) and elastic horizontal scalability, dynamically managed using the REST API.
• State via Kafka Topics — Distributed Mode consistency is fully maintained through three internal topics: connect-configs (1 partition), connect-offsets, and connect-status.
• Cooperative Rebalancing — The modern Connect protocol minimizes latency disruptions by only moving failure-affected tasks without stopping other tasks’ processing in the cluster.
• Replication Factor — Connect’s internal topics must be set with a minimum replication factor of 3 in production to protect data pipeline configuration integrity from broker failures.

← Previous: Source vs Sink Connector Next: Database Integration & CDC →