Connector configuration in Kafka - Commands & Configuration

Introduction

Connectors help move data between Kafka and other systems automatically. Connector configuration files tell Kafka how to connect and transfer data without manual coding.

When you want to import data from a database into Kafka automatically.

When you need to export Kafka messages to a file system or cloud storage.

When you want to stream data from an API into Kafka topics.

When you want to move data from Kafka to another messaging system.

When you want to automate data syncing between Kafka and external systems.

Config File - jdbc-source-connector.json

jdbc-source-connector.json

{
  "name": "jdbc-source-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSourceConnector",
    "tasks.max": "1",
    "connection.url": "jdbc:postgresql://localhost:5432/mydb",
    "connection.user": "myuser",
    "connection.password": "mypassword",
    "table.whitelist": "customers",
    "mode": "incrementing",
    "incrementing.column.name": "id",
    "topic.prefix": "jdbc-",
    "poll.interval.ms": "5000"
  }
}

This JSON configures a Kafka JDBC source connector.

name: Unique connector name.

connector.class: The connector plugin to use.

tasks.max: Number of parallel tasks.

connection.url/user/password: Database connection details.

table.whitelist: Which table to read from.

mode: How to detect new rows (incrementing id).

topic.prefix: Prefix for Kafka topics created.

poll.interval.ms: How often to check for new data.

Commands

This command creates the connector by sending the configuration file to Kafka Connect's REST API.

Terminal

curl -X POST -H "Content-Type: application/json" --data @jdbc-source-connector.json http://localhost:8083/connectors

Expected Output

{"name":"jdbc-source-connector","config":{"connector.class":"io.confluent.connect.jdbc.JdbcSourceConnector","tasks.max":"1","connection.url":"jdbc:postgresql://localhost:5432/mydb","connection.user":"myuser","connection.password":"mypassword","table.whitelist":"customers","mode":"incrementing","incrementing.column.name":"id","topic.prefix":"jdbc-","poll.interval.ms":"5000"},"tasks":[],"type":"source"}

→

-X POST - Send a POST request to create the connector

→

-H "Content-Type: application/json" - Tell the server the data is JSON

→

--data @jdbc-source-connector.json - Send the connector config file as data

Check the status of the connector to see if it is running and tasks are active.

Terminal

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

Expected Output

{"name":"jdbc-source-connector","connector":{"state":"RUNNING","worker_id":"worker1:8083"},"tasks":[{"id":0,"state":"RUNNING","worker_id":"worker1:8083"}],"type":"source"}

Consume the first 5 messages from the Kafka topic created by the connector to verify data is flowing.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic jdbc-customers --from-beginning --max-messages 5

Expected Output

{"id":1,"name":"Alice","email":"[e-mail]"}

{"id":2,"name":"Bob","email":"[e-mail]"}

{"id":3,"name":"Carol","email":"[e-mail]"}

{"id":4,"name":"Dave","email":"[e-mail]"}

{"id":5,"name":"Eve","email":"[e-mail]"}

→

--from-beginning - Read messages from the start of the topic

→

--max-messages 5 - Stop after reading 5 messages

Key Concept

Connector configuration files tell Kafka Connect how to move data automatically between Kafka and other systems without manual coding.

Common Mistakes

Using incorrect database connection URL or credentials

Connector cannot connect to the database, so no data is imported

Double-check and test the database URL, username, and password before applying the config

Not setting the correct mode (e.g., incrementing or timestamp) for detecting new data

Connector may miss new rows or duplicate data in Kafka topics

Choose the mode that matches your database schema and data update pattern

Not verifying connector status after creation

You might think the connector is running when it has errors or is stopped

Always check connector status with the REST API to confirm it is running

Summary

Create a JSON configuration file that defines how the connector moves data.

Use curl to POST the config to Kafka Connect and create the connector.

Check the connector status to ensure it is running properly.

Consume messages from the Kafka topic to verify data flow.

Practice

1. What is the main purpose of a connector configuration in Kafka Connect?

easy

A. To specify how data moves between Kafka and external systems

B. To define the Kafka broker's internal settings

C. To configure Kafka topic partitions

D. To set the Kafka consumer group ID

Solution

1. Step 1: Understand Kafka Connect role

   Kafka Connect uses connectors to move data between Kafka and other systems.

2. Step 2: Identify configuration purpose

   The connector configuration tells Kafka Connect how to perform this data movement.

3. Final Answer:

   To specify how data moves between Kafka and external systems -> Option A

4. Quick Check:

   Connector configuration = data movement setup [OK]

Hint: Connector config defines data flow direction [OK]

Common Mistakes:

• Confusing connector config with Kafka broker settings
• Thinking it configures Kafka topic partitions
• Mixing up consumer group ID with connector config

2. Which of the following is the correct way to specify the connector class in a Kafka Connect configuration JSON?

easy

A. "connectorClassName": "org.apache.kafka.connect.file.FileStreamSourceConnector"

B. "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector"

C. "connectorClass": "org.apache.kafka.connect.file.FileStreamSourceConnector"

D. "connector-class": "org.apache.kafka.connect.file.FileStreamSourceConnector"

Solution

1. Step 1: Recall Kafka Connect JSON key format

   Kafka Connect uses keys with dot notation, e.g., "connector.class".

2. Step 2: Match correct key name

   Only "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector" uses the correct key "connector.class" with quotes and colon.

3. Final Answer:

   "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector" -> Option B

4. Quick Check:

   Correct key is "connector.class" [OK]

Hint: Use dot notation keys like "connector.class" [OK]

Common Mistakes:

• Using hyphens instead of dots in keys
• Using camelCase keys instead of dot notation
• Omitting quotes around keys or values

3. Given this connector configuration snippet:

{
  "name": "my-source-connector",
  "connector.class": "org.apache.kafka.connect.file.FileStreamSourceConnector",
  "tasks.max": "2",
  "file": "/tmp/input.txt",
  "topic": "my-topic"
}

What does the tasks.max setting control?

medium

A. Maximum number of Kafka topics to consume

B. Maximum number of retries for failed tasks

C. Maximum size of each task's data batch

D. Maximum number of tasks to run in parallel for this connector

Solution

1. Step 1: Understand tasks.max meaning

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

2. Step 2: Eliminate other options

   It does not control topic count, batch size, or retry count.

3. Final Answer:

   Maximum number of tasks to run in parallel for this connector -> Option D

4. Quick Check:

   tasks.max = parallel task limit [OK]

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

Common Mistakes:

• Confusing tasks.max with topic count
• Thinking it controls batch size
• Assuming it sets retry attempts

4. You have this connector config JSON:

{
  "name": "my-sink-connector",
  "connector.class": "org.apache.kafka.connect.file.FileStreamSinkConnector",
  "tasks.max": 1,
  "topics": "my-topic"
}

What is wrong with this configuration?

medium

A. "connector.class" value is incorrect for a sink connector

B. "topics" key should be "topic" (singular)

C. Missing "file" property to specify output file

D. "tasks.max" should be a string, not a number

Solution

1. Step 1: Check required properties for FileStreamSinkConnector

   This connector needs a "file" property to specify the output file path.

2. Step 2: Validate other properties

   "tasks.max" can be a number or string, "topics" is valid plural, and "connector.class" is correct.

3. Final Answer:

   Missing "file" property to specify output file -> Option C

4. Quick Check:

   FileStreamSinkConnector requires "file" property [OK]

Hint: FileStreamSinkConnector needs "file" path set [OK]

Common Mistakes:

• Assuming "tasks.max" must be string
• Confusing "topic" and "topics" keys
• Changing connector.class unnecessarily

5. You want to configure a Kafka Connect source connector to read from a database table and write to a Kafka topic. Which configuration property is essential to specify the Kafka topic where data will be published?

hard

A. "topic"

B. "topics"

C. "topics.regex"

D. "kafka.topic"

Solution

1. Step 1: Understand topic properties in connectors

   Source connectors typically use "topic" (singular) to specify the Kafka topic to publish data to.

2. Step 2: Differentiate from other keys

   "topics" is common for sink connectors; "topics.regex" is for pattern matching; "kafka.topic" is not standard.

3. Final Answer:

   "topic" -> Option A

4. Quick Check:

   Use "topic" to specify Kafka topic for sources [OK]

Hint: Use "topic" for source connector Kafka topic [OK]

Common Mistakes:

• Using plural "topics" instead of "topic"
• Confusing "topics.regex" with direct topic specification
• Inventing non-standard keys like "kafka.topic"