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.