Kafka Connect architecture - Commands & Configuration

Introduction

Kafka Connect helps move data between Kafka and other systems automatically. It solves the problem of writing custom code to connect databases, files, or other services to Kafka.

When you want to copy data from a database into Kafka without writing code

When you need to export Kafka data to a search engine like Elasticsearch

When you want to stream data from files or logs into Kafka topics

When you want to keep data synchronized between Kafka and external systems

When you want to run connectors in a scalable and fault-tolerant way

Config File - connect-distributed.properties

connect-distributed.properties

bootstrap.servers=localhost: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
config.storage.topic=connect-configs
offset.storage.topic=connect-offsets
status.storage.topic=connect-status
config.storage.replication.factor=1
offset.storage.replication.factor=1
status.storage.replication.factor=1
rest.port=8083

This file configures Kafka Connect in distributed mode.

• bootstrap.servers: Kafka brokers to connect to.
• key.converter and value.converter: How data is serialized.
• config.storage.topic, offset.storage.topic, status.storage.topic: Internal topics to store connector configs, offsets, and status.
• replication.factor: Number of copies for fault tolerance.
• rest.port: Port for REST API to manage connectors.

Commands

Starts Kafka Connect in distributed mode using the configuration file. This runs the service that manages connectors and tasks.

Terminal

connect-distributed.sh connect-distributed.properties

Expected Output

[2024-06-01 12:00:00,000] INFO Kafka Connect distributed worker started (org.apache.kafka.connect.runtime.ConnectDistributed)

Creates a new source connector that reads from a file and writes to a Kafka topic. This shows how to add connectors via REST API.

Terminal

curl -X POST -H "Content-Type: application/json" --data '{"name": "my-source-connector", "config": {"connector.class": "FileStreamSource", "tasks.max": "1", "file": "/tmp/input.txt", "topic": "my-topic"}}' http://localhost:8083/connectors

Expected Output

{"name":"my-source-connector","config":{"connector.class":"FileStreamSource","tasks.max":"1","file":"/tmp/input.txt","topic":"my-topic"},"tasks":[],"type":"source"}

Lists all connectors currently running in Kafka Connect. This verifies the connector was created successfully.

Terminal

curl http://localhost:8083/connectors

Expected Output

["my-source-connector"]

Shows the status of the specific connector and its tasks to check if it is running properly.

Terminal

curl http://localhost:8083/connectors/my-source-connector/status

Expected Output

{"name":"my-source-connector","connector":{"state":"RUNNING","worker_id":"connect-worker-1"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"connect-worker-1"}]}

Key Concept

If you remember nothing else from this pattern, remember: Kafka Connect runs connectors as separate tasks managed by a distributed service that handles data movement automatically.

Common Mistakes

Not setting the internal storage topics in the config file

Kafka Connect needs these topics to store connector configs, offsets, and status. Without them, it cannot track progress or recover.

Always define config.storage.topic, offset.storage.topic, and status.storage.topic with proper replication factors.

Starting Kafka Connect without Kafka brokers running

Kafka Connect depends on Kafka brokers to send and receive data. Without brokers, it will fail to start or connect.

Make sure Kafka brokers are running and reachable at the bootstrap.servers address before starting Kafka Connect.

Using incompatible converters for key and value

If converters do not match the data format, connectors will fail to serialize or deserialize data correctly.

Use matching converters like JsonConverter for both key and value or configure them according to your data format.

Summary

Kafka Connect runs as a distributed service managing connectors and tasks to move data automatically.

You configure Kafka Connect with a properties file specifying Kafka brokers, converters, and internal topics.

Connectors are created and managed via REST API calls to add, list, and check status.

Practice

1. What is the main role of a Kafka Connect worker in the Kafka Connect architecture?

easy

A. To run connectors and tasks that move data between Kafka and external systems

B. To store Kafka topic data permanently

C. To produce messages to Kafka topics

D. To manage Kafka broker configurations

Solution

1. Step 1: Understand Kafka Connect components

   Kafka Connect architecture includes workers, connectors, and tasks. Workers run connectors and tasks.

2. Step 2: Identify worker's role

   Workers execute connectors and tasks to move data between Kafka and external systems.

3. Final Answer:

   To run connectors and tasks that move data between Kafka and external systems -> Option A

4. Quick Check:

   Worker = runs connectors/tasks [OK]

Hint: Workers run connectors and tasks to move data [OK]

Common Mistakes:

• Confusing workers with Kafka brokers
• Thinking workers store data permanently
• Mixing up producers and workers

2. Which of the following is the correct way to start Kafka Connect in standalone mode using the command line?

easy

A. connect-distributed.sh worker.properties connector.properties

B. connect-standalone.sh worker.properties connector.properties

C. kafka-server-start.sh connect-standalone.properties

D. kafka-connect-start.sh standalone.properties

Solution

1. Step 1: Recall Kafka Connect startup commands

   Standalone mode uses connect-standalone.sh with worker and connector config files.

2. Step 2: Match correct command syntax

   connect-standalone.sh worker.properties connector.properties matches the correct command format: connect-standalone.sh worker.properties connector.properties

3. Final Answer:

   connect-standalone.sh worker.properties connector.properties -> Option B

4. Quick Check:

   Standalone mode uses connect-standalone.sh [OK]

Hint: Standalone mode uses connect-standalone.sh command [OK]

Common Mistakes:

• Using connect-distributed.sh for standalone mode
• Confusing Kafka broker start scripts with Connect scripts
• Wrong file names or missing config files

3. Given the following Kafka Connect distributed mode configuration snippet:

{
  "name": "my-source-connector",
  "config": {
    "connector.class": "FileStreamSource",
    "tasks.max": "2",
    "file": "/tmp/input.txt",
    "topic": "test-topic"
  }
}

What does the tasks.max setting control?

medium

A. The maximum number of tasks to run for this connector

B. The maximum number of Kafka brokers to connect

C. The maximum size of each message sent

D. The maximum number of topics to read from

Solution

1. Step 1: Understand the tasks.max parameter

   In Kafka Connect, tasks.max sets how many tasks a connector can run in parallel.

2. Step 2: Apply to given config

   Here, tasks.max is 2, so the connector can run up to 2 tasks to increase throughput.

3. Final Answer:

   The maximum number of tasks to run for this connector -> Option A

4. Quick Check:

   tasks.max = max tasks per connector [OK]

Hint: tasks.max limits parallel tasks for connector [OK]

Common Mistakes:

• Confusing tasks.max with broker count
• Thinking it limits message size
• Assuming it controls topic count

4. You have a distributed Kafka Connect cluster but notice that connectors are not starting tasks as expected. Which of the following is the most likely cause?

medium

A. The connector configuration file is missing the tasks.max property

B. Kafka brokers are offline

C. The worker.properties file is empty

D. The connector class name is misspelled in the configuration

Solution

1. Step 1: Analyze why tasks don't start

   If connectors don't start tasks, a common cause is an invalid connector class name.

2. Step 2: Check other options

   Missing tasks.max defaults to 1 task, so not likely cause. Brokers offline cause other errors. Empty worker.properties would prevent worker start. Misspelled connector class prevents task creation.

3. Final Answer:

   The connector class name is misspelled in the configuration -> Option D

4. Quick Check:

   Misspelled connector class blocks tasks [OK]

Hint: Check connector class spelling if tasks don't start [OK]

Common Mistakes:

• Assuming missing tasks.max stops tasks
• Ignoring connector class name correctness
• Not verifying worker config completeness

5. You want to build a fault-tolerant data pipeline using Kafka Connect that can scale horizontally and recover from worker failures. Which Kafka Connect mode should you choose and why?

hard

A. Standalone mode, because it is simpler and easier to configure

B. Distributed mode, because it requires no configuration files

C. Distributed mode, because it supports multiple workers and fault tolerance

D. Standalone mode, because it automatically balances tasks across workers

Solution

1. Step 1: Understand Kafka Connect modes

   Standalone mode runs a single worker, no fault tolerance or scaling. Distributed mode runs multiple workers, supports scaling and fault tolerance.

2. Step 2: Match requirements to mode

   For fault tolerance and horizontal scaling, distributed mode is required.

3. Final Answer:

   Distributed mode, because it supports multiple workers and fault tolerance -> Option C

4. Quick Check:

   Fault tolerance needs distributed mode [OK]

Hint: Use distributed mode for scaling and fault tolerance [OK]

Common Mistakes:

• Choosing standalone for fault tolerance
• Thinking standalone balances tasks
• Believing distributed mode needs no config