Sink connectors in Kafka - Step-by-Step Execution

Process Flow - Sink connectors

Kafka Topic receives messages

↓

Sink Connector polls messages

↓

Transforms/Processes messages

↓

Writes messages to external system

↓

Acknowledges success or retries on failure

Sink connectors read messages from Kafka topics and write them to external systems, handling processing and delivery.

Execution Sample

Kafka

name=jdbc-sink
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
tasks.max=1
topics=my_topic
connection.url=jdbc:postgresql://localhost:5432/mydb
insert.mode=insert

This config sets up a JDBC sink connector to write Kafka topic 'my_topic' data into a PostgreSQL database.

Process Table

Step	Action	Input	Output	Notes
1	Kafka topic receives message	{"id":1,"name":"Alice"}	Message stored in topic	Message produced by producer
2	Sink connector polls topic	Message available in topic	Message fetched by connector	Connector reads new message
3	Process message	{"id":1,"name":"Alice"}	Transformed record	May apply schema or conversion
4	Write to external system	Transformed record	Record inserted in DB	Writes to PostgreSQL table
5	Acknowledge success	Record inserted	Commit offset	Connector commits offset to avoid reprocessing
6	Poll next message	Waiting for new messages	Idle or next fetch	Connector waits or fetches next batch
7	No more messages	Topic empty	Stop polling	Connector pauses until new data arrives

💡 No more messages in topic, connector waits for new data

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	After Step 5	Final
message	null	{"id":1,"name":"Alice"}	Transformed record	Written to DB	Committed offset	null (ready for next)
offset	0	0	0	0	1	1

Key Moments - 3 Insights

Why does the connector commit the offset after writing to the external system?

Committing the offset after writing ensures messages are not lost or duplicated. See execution_table step 5 where offset commit happens only after successful write.

What happens if the external system is down when the connector tries to write?

The connector retries writing and does not commit the offset until successful, preventing data loss. This is implied between steps 4 and 5 in the execution_table.

Does the sink connector modify the original Kafka message?

The connector may transform the message format but does not change the original Kafka message in the topic. See step 3 where processing happens separately.

Visual Quiz - 3 Questions

Look at the execution_table, what is the connector doing at step 3?

A Polling the Kafka topic for new messages

B Processing or transforming the fetched message

C Writing the message to the external system

D Committing the offset after successful write

💡 Hint

Refer to execution_table row 3 under 'Action' and 'Output'

At which step does the connector commit the offset to avoid reprocessing?

A Step 5

B Step 4

C Step 2

D Step 6

💡 Hint

Check execution_table step 5 notes about offset commit

If the external system is unavailable, what will the connector do according to the flow?

A Skip the message and commit offset

B Delete the message from Kafka topic

C Retry writing and not commit offset

D Stop polling new messages immediately

💡 Hint

See key_moments explanation about retries and offset commit

Concept Snapshot

Sink connectors read messages from Kafka topics.
They transform and write data to external systems.
Offset commits happen only after successful writes.
Retries occur if external systems fail.
This ensures no data loss or duplication.

Full Transcript

Sink connectors in Kafka take messages from topics and send them to external systems like databases. The connector polls the topic, processes messages, writes them out, and commits offsets only after successful delivery. This prevents losing or duplicating data. If the external system is down, the connector retries writing and waits before committing. This flow ensures reliable data transfer from Kafka to other systems.

Practice

1.

What is the main purpose of a Kafka sink connector?

easy

A. To create new Kafka topics dynamically

B. To move data from external systems into Kafka topics

C. To monitor Kafka cluster health

D. To move data from Kafka topics to external systems automatically

Solution

1. Step 1: Understand the role of sink connectors

   Sink connectors are designed to export data from Kafka topics to other systems.

2. Step 2: Differentiate from source connectors

   Source connectors import data into Kafka, while sink connectors export data out.

3. Final Answer:

   To move data from Kafka topics to external systems automatically -> Option D

4. Quick Check:

   Sink connector = export data from Kafka [OK]

Hint: Sink connectors export data out of Kafka topics [OK]

Common Mistakes:

• Confusing sink with source connectors
• Thinking sink connectors create topics
• Assuming sink connectors monitor Kafka health

2.

Which of the following is a correct JSON field to specify the Kafka topic for a sink connector configuration?

{
  "name": "my-sink-connector",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "my_topic",
    "connection.url": "jdbc:mysql://localhost:3306/mydb"
  }
}

easy

A. "topics"

B. "topic"

C. "topic_name"

D. "source.topics"

Solution

1. Step 1: Check the standard Kafka Connect config field

   The field to specify Kafka topics in sink connector config is "topics" (plural).

2. Step 2: Verify the example JSON

   The example uses "topics": "my_topic", which is correct syntax.

3. Final Answer:

   "topics" -> Option A

4. Quick Check:

   Kafka sink config uses "topics" field [OK]

Hint: Use "topics" (plural) to list Kafka topics in config [OK]

Common Mistakes:

• Using "topic" (singular) instead of "topics"
• Using non-standard fields like "topic_name"
• Confusing source connector fields with sink

3.

Given this sink connector config snippet, what will be the destination system?

{
  "name": "file-sink-connector",
  "config": {
    "connector.class": "FileStreamSink",
    "topics": "logs",
    "file": "/tmp/output.log"
  }
}

medium

A. A Kafka topic named 'logs'

B. A file located at /tmp/output.log

C. A database table named 'logs'

D. A remote HTTP endpoint

Solution

1. Step 1: Identify the connector class

   The connector class is "FileStreamSink", which writes data to a file.

2. Step 2: Check the "file" config field

   The "file" field specifies the destination file path "/tmp/output.log".

3. Final Answer:

   A file located at /tmp/output.log -> Option B

4. Quick Check:

   FileStreamSink writes to file path [OK]

Hint: FileStreamSink writes Kafka data to a file path [OK]

Common Mistakes:

• Assuming data goes back to Kafka topic
• Confusing sink with source connectors
• Thinking it writes to a database or HTTP

4.

Identify the error in this sink connector config JSON:

{
  "name": "jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topic": "orders",
    "connection.url": "jdbc:postgresql://localhost:5432/shop"
  }
}

medium

A. Invalid JDBC URL format

B. Connector class name is incorrect

C. Missing 'topics' field; uses 'topic' instead

D. Missing 'name' field

Solution

1. Step 1: Check the topic field name

   The config uses "topic" (singular), but Kafka sink connectors require "topics" (plural).

2. Step 2: Verify other fields

   The JDBC URL and connector class are correctly formatted, and "name" is present.

3. Final Answer:

   Missing 'topics' field; uses 'topic' instead -> Option C

4. Quick Check:

   Use "topics" not "topic" in sink config [OK]

Hint: Use plural "topics" field in sink connector config [OK]

Common Mistakes:

• Using singular "topic" instead of "topics"
• Assuming JDBC URL is wrong without checking
• Thinking connector class name is invalid

5.

You want to export data from multiple Kafka topics to a single database table using a JDBC sink connector. Which configuration approach is correct?

{
  "name": "multi-topic-jdbc-sink",
  "config": {
    "connector.class": "io.confluent.connect.jdbc.JdbcSinkConnector",
    "topics": "topic1,topic2,topic3",
    "connection.url": "jdbc:mysql://localhost:3306/mydb",
    "table.name.format": "combined_table"
  }
}

hard

A. This config exports all topics to one table named 'combined_table'

B. Each topic will create a separate table ignoring 'table.name.format'

C. The connector will fail because multiple topics are not allowed

D. Data will be merged but only from 'topic1'

Solution

1. Step 1: Understand 'topics' field with multiple topics

   The "topics" field lists multiple topics separated by commas, which is valid.

2. Step 2: Check 'table.name.format' usage

   Setting "table.name.format" to a fixed table name causes all data to go into that single table.

3. Step 3: Confirm connector behavior

   The JDBC sink connector supports writing multiple topics to one table if configured this way.

4. Final Answer:

   This config exports all topics to one table named 'combined_table' -> Option A

5. Quick Check:

   Multiple topics + fixed table name = single table export [OK]

Hint: Use comma-separated topics and fixed table name for single table export [OK]

Common Mistakes:

• Assuming multiple topics create separate tables automatically
• Thinking multiple topics are not allowed in sink config
• Ignoring 'table.name.format' effect