Source connectors in Kafka - Commands & Configuration

Introduction

Source connectors help move data from external systems into Kafka automatically. They solve the problem of manually writing code to import data by providing ready-made tools to connect databases, files, or other sources to Kafka topics.

When you want to stream data from a database into Kafka for real-time processing.

When you need to import log files continuously into Kafka for analysis.

When you want to capture changes from a source system and send them to Kafka topics.

When you want to integrate external data sources with Kafka without custom coding.

When you want to automate data ingestion pipelines into Kafka.

Config File - source-connector.json

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": "mydb-",
    "poll.interval.ms": "5000"
  }
}

This JSON configures a JDBC source connector to import data from a PostgreSQL database.

• name: Unique name for the connector.
• connector.class: The connector plugin to use (JDBC source).
• tasks.max: Number of parallel tasks to run.
• connection.url, connection.user, connection.password: Database connection details.
• table.whitelist: Which tables to import data from.
• mode: How to detect new rows (incrementing column).
• incrementing.column.name: Column used to track new rows.
• topic.prefix: Prefix added to Kafka topic names.
• poll.interval.ms: How often to check for new data.

Commands

This command creates the source connector by sending the JSON config to Kafka Connect's REST API.

Terminal

curl -X POST -H "Content-Type: application/json" --data @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":"mydb-","poll.interval.ms":"5000"},"tasks":[],"type":"source"}

→

-X POST - Specifies the HTTP method to create the connector

→

-H "Content-Type: application/json" - Sets the content type header for JSON data

→

--data @source-connector.json - Sends the connector configuration from the file

This command checks the status of the source connector to verify it is running.

Terminal

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

Expected Output

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

This command reads the first 5 messages from the Kafka topic where the source connector sends data, to verify data ingestion.

Terminal

kafka-console-consumer --bootstrap-server localhost:9092 --topic mydb-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 - Reads messages from the start of the topic

→

--max-messages 5 - Limits output to 5 messages

Key Concept

If you remember nothing else from this pattern, remember: source connectors automate importing data from external systems into Kafka topics without custom coding.

Common Mistakes

Using incorrect database connection details in the connector config.

The connector cannot connect to the source database and fails to start.

Double-check and test the database URL, username, and password before creating the connector.

Not specifying the correct mode and incrementing column for incremental data import.

The connector may import duplicate data or miss new rows.

Set mode to 'incrementing' or 'timestamp+incrementing' and specify the correct column to track new data.

Not verifying connector status after creation.

You may think the connector is running when it has failed or is paused.

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

Summary

Create a source connector by posting a JSON config to Kafka Connect REST API.

Check the connector status to ensure it is running properly.

Consume messages from the Kafka topic to verify data is flowing from the source system.

Practice

1. What is the main purpose of a Kafka source connector?

easy

A. To automatically import data from external systems into Kafka topics

B. To export data from Kafka topics to external systems

C. To monitor Kafka cluster health

D. To manage Kafka topic partitions

Solution

1. Step 1: Understand the role of source connectors

   Source connectors are designed to bring data from outside systems into Kafka topics automatically.

2. Step 2: Differentiate from sink connectors

   Sink connectors do the opposite by exporting data from Kafka to external systems, so they are not the correct answer here.

3. Final Answer:

   To automatically import data from external systems into Kafka topics -> Option A

4. Quick Check:

   Source connector = import data [OK]

Hint: Source connectors bring data into Kafka topics [OK]

Common Mistakes:

• Confusing source connectors with sink connectors
• Thinking connectors manage Kafka internals like partitions
• Assuming connectors monitor Kafka health

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

easy

A. "sourceConnector"

B. "source.class"

C. "connectorType"

D. "connector.class"

Solution

1. Step 1: Recall Kafka Connect configuration syntax

   The key to specify the connector implementation class is always "connector.class" in the JSON config.

2. Step 2: Eliminate incorrect options

   Other options like "source.class", "connectorType", or "sourceConnector" are not valid Kafka Connect configuration keys.

3. Final Answer:

   "connector.class" -> Option D

4. Quick Check:

   Connector class key = "connector.class" [OK]

Hint: Use "connector.class" to specify connector type [OK]

Common Mistakes:

• Using incorrect or made-up JSON keys
• Confusing source connector config keys with sink connector keys
• Omitting quotes around JSON keys

3. Given this Kafka Connect source connector configuration snippet:

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

What will happen when this connector runs?

medium

A. It fails because 'tasks.max' must be an integer, not a string

B. It reads data from /tmp/input.txt and writes it to the Kafka topic 'test-topic'

C. It creates a new Kafka topic named '/tmp/input.txt'

D. It writes data from Kafka topic 'test-topic' to /tmp/input.txt

Solution

1. Step 1: Identify connector type and purpose

   The "FileStreamSource" connector reads data from a file and sends it to a Kafka topic.

2. Step 2: Understand configuration keys

   "file" specifies the input file, "topic" specifies the Kafka topic to write to, and "tasks.max" as a string is acceptable in JSON configs.

3. Final Answer:

   It reads data from /tmp/input.txt and writes it to the Kafka topic 'test-topic' -> Option B

4. Quick Check:

   FileStreamSource reads file to topic [OK]

Hint: FileStreamSource reads file into Kafka topic [OK]

Common Mistakes:

• Confusing source and sink connectors
• Thinking 'tasks.max' must be integer type (string is valid in JSON)
• Assuming connector creates topics automatically

4. You configured a source connector with this JSON:

{
  "name": "my-connector",
  "connector.class": "JdbcSourceConnector",
  "tasks.max": 2,
  "connection.url": "jdbc:mysql://localhost:3306/mydb",
  "topic.prefix": "db-"
}

But the connector fails to start. What is the most likely cause?

medium

A. Missing required property 'mode' for JdbcSourceConnector

B. Incorrect 'tasks.max' value; must be a string

C. The 'topic.prefix' should be 'topic.name' instead

D. The 'connection.url' must use PostgreSQL syntax

Solution

1. Step 1: Review JdbcSourceConnector required properties

   JdbcSourceConnector requires a 'mode' property to specify how to query the database (e.g., 'incrementing', 'timestamp+incrementing').

2. Step 2: Check other properties

   'tasks.max' as integer is valid, 'topic.prefix' is correct, and 'connection.url' can be MySQL syntax if using MySQL.

3. Final Answer:

   Missing required property 'mode' for JdbcSourceConnector -> Option A

4. Quick Check:

   JdbcSourceConnector needs 'mode' property [OK]

Hint: JdbcSourceConnector requires 'mode' property [OK]

Common Mistakes:

• Assuming 'tasks.max' must be string
• Confusing 'topic.prefix' with 'topic.name'
• Thinking connection URL must be PostgreSQL always

5. You want to ingest data from multiple CSV files located in a directory into Kafka using a source connector. Which approach best applies?

hard

A. Use JdbcSourceConnector with CSV files as database tables

B. Use the FileStreamSource connector with the directory path as 'file' property

C. Use a FilePulse source connector configured with the directory path to read all CSV files

D. Manually write a Kafka producer to read files and send data

Solution

1. Step 1: Identify connector capabilities

   FilePulse source connector supports reading multiple files from a directory and processing CSV files.

2. Step 2: Evaluate other options

   FileStreamSource reads only a single file, JdbcSourceConnector is for databases, and manual coding is not using source connectors.

3. Final Answer:

   Use a FilePulse source connector configured with the directory path to read all CSV files -> Option C

4. Quick Check:

   FilePulse reads multiple files from directory [OK]

Hint: FilePulse connector handles multiple files ingestion [OK]

Common Mistakes:

• Using FileStreamSource for multiple files
• Trying to use JdbcSourceConnector for CSV files
• Ignoring built-in connectors and coding manually