Source connectors in Kafka - Step-by-Step Execution

Process Flow - Source connectors

Start Connector

↓

Connect to Source System

↓

Read Data from Source

↓

Convert Data to Kafka Format

↓

Send Data to Kafka Topic

↓

Repeat Reading Continuously

↓

Stop Connector

The source connector starts, connects to the source system, reads data, converts it, sends it to Kafka, and repeats until stopped.

Execution Sample

Kafka

name=source-connector
connector.class=FileStreamSourceConnector
tasks.max=1
file=/tmp/input.txt
topic=source-topic

This config sets up a source connector that reads from a file and sends data to a Kafka topic.

Process Table

Step	Action	Source System State	Connector State	Kafka Topic State
1	Start connector	File /tmp/input.txt exists	Connector initialized	Topic empty
2	Connect to source	File accessible	Connected to file	Topic empty
3	Read line 1	File has lines	Read first line	Topic empty
4	Convert data	Line 1 read	Data converted	Topic empty
5	Send to topic	Line 1 read	Data sent	Topic has 1 message
6	Read line 2	File has more lines	Read second line	Topic has 1 message
7	Convert data	Line 2 read	Data converted	Topic has 1 message
8	Send to topic	Line 2 read	Data sent	Topic has 2 messages
9	No more lines	End of file	Waiting for new data	Topic has 2 messages
10	Stop connector	File unchanged	Connector stopped	Topic has 2 messages

💡 Connector stops when manually stopped or on error.

Status Tracker

Variable	Start	After Step 3	After Step 5	After Step 8	Final
Connector State	Not started	Reading line 1	Sent line 1	Sent line 2	Stopped
Kafka Topic Messages	0	0	1	2	2
Source File Lines Read	0	1	1	2	2

Key Moments - 2 Insights

Why does the connector keep running after reading all current lines?

Because source connectors continuously poll the source for new data, as shown in step 9 where it waits for new data after reaching end of file.

What happens if the source file is not accessible at start?

The connector fails to connect and does not proceed, as indicated in step 2 where connection to source is required before reading.

Visual Quiz - 3 Questions

Look at the execution table, how many messages are in the Kafka topic after step 5?

A 1

B 0

C 2

D None

💡 Hint

Check the 'Kafka Topic State' column at step 5 in the execution table.

At which step does the connector read the second line from the source file?

A Step 3

B Step 6

C Step 8

D Step 9

💡 Hint

Look at the 'Action' column for reading lines in the execution table.

If the source file had no lines, what would the connector state be after step 3?

A Reading line 1

B Connector stopped

C Waiting for new data

D Data sent

💡 Hint

Refer to the connector behavior when no data is available in the execution table and variable tracker.

Concept Snapshot

Source connectors read data from external systems
Convert data to Kafka format
Send data continuously to Kafka topics
Keep running to capture new data
Configured via connector properties

Full Transcript

Source connectors in Kafka start by connecting to an external source system like a file or database. They read data from that source, convert it into Kafka's message format, and send it to a Kafka topic. This process repeats continuously to capture new data as it arrives. The connector runs until it is stopped manually or encounters an error. The example configuration shows a file source connector reading lines from a file and sending them to a Kafka topic. The execution table traces each step: starting, connecting, reading lines, converting, sending, waiting for new data, and stopping. Variables like connector state, messages in the topic, and lines read from the source change step by step. Key points include understanding that connectors keep running to capture new data and that they must connect successfully to the source before reading. The visual quiz tests understanding of message counts, reading steps, and connector states when no data is present.

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