Common connectors (JDBC, S3, Elasticsearch) in Kafka - Step-by-Step Execution

Process Flow - Common connectors (JDBC, S3, Elasticsearch)

Kafka Source Topic

↓

Connector Configuration

↓

Connector Plugin

↓

JDBC

↓

Data Sink

Data flows from Kafka through a connector configured to send data to JDBC, S3, or Elasticsearch sinks.

Execution Sample

Kafka

name=jdbc-sink-connector
connector.class=io.confluent.connect.jdbc.JdbcSinkConnector
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 data into a PostgreSQL database.

Process Table

Step	Action	Connector	Input	Output	Result
1	Read message from Kafka topic	All	Kafka message	Message data	Message ready for processing
2	Apply connector config	JDBC	Message data	SQL Insert statement	Prepared SQL for DB
3	Send data to sink	JDBC	SQL Insert statement	Database row	Data stored in DB
4	Read message from Kafka topic	S3	Kafka message	Message data	Message ready for processing
5	Apply connector config	S3	Message data	File object	Data formatted for S3
6	Send data to sink	S3	File object	S3 bucket file	Data stored in S3
7	Read message from Kafka topic	Elasticsearch	Kafka message	Message data	Message ready for processing
8	Apply connector config	Elasticsearch	Message data	JSON document	Data formatted for ES
9	Send data to sink	Elasticsearch	JSON document	ES index document	Data indexed in ES
10	No more messages	All	-	-	Connector idle, waiting for new messages

💡 No more messages in Kafka topic, connectors wait for new data.

Status Tracker

Variable	Start	After Step 1	After Step 2	After Step 3	After Step 4	After Step 5	After Step 6	After Step 7	After Step 8	After Step 9	Final
Message	None	Kafka message	Message data	SQL Insert statement	None	Kafka message	File object	None	Kafka message	JSON document	None
Data Sink	Empty	Empty	Empty	Database row	Empty	Empty	S3 bucket file	Empty	Empty	ES index document	Idle

Key Moments - 3 Insights

Why does the message variable reset to None after sending to the sink?

After the connector sends data to the sink (see steps 3, 6, 9), the message is processed and cleared to prepare for the next message, as shown in the execution_table.

How does the connector know which sink to send data to?

The connector.class setting in the configuration (see execution_sample) determines which sink plugin is used, directing data to JDBC, S3, or Elasticsearch accordingly.

Why are there repeated steps for reading messages in the execution_table?

Each connector independently reads messages from Kafka to process them for its sink, so steps repeat for JDBC, S3, and Elasticsearch connectors separately.

Visual Quiz - 3 Questions

Look at the execution_table, what is the output after step 5?

A SQL Insert statement

B File object

C JSON document

D Database row

💡 Hint

Check the 'Output' column for step 5 in the execution_table.

At which step does the data get stored in Elasticsearch?

A Step 9

B Step 6

C Step 3

D Step 10

💡 Hint

Look for 'Data indexed in ES' in the 'Result' column of the execution_table.

If the connector.class is changed to S3, which steps would no longer occur?

A Steps 4 and 5

B Steps 7 and 8

C Steps 2 and 3

D Steps 1 and 10

💡 Hint

Refer to the 'Connector' column in the execution_table to see which steps belong to JDBC.

Concept Snapshot

Common Kafka connectors move data from Kafka topics to external systems.
Use connector.class to select sink: JDBC for databases, S3 for storage, Elasticsearch for search.
Connectors read messages, transform data, then write to the sink.
Each connector runs independently but follows the same flow: read, transform, write.
Configure connectors with properties like topics, connection URLs, and modes.

Full Transcript

This visual execution shows how Kafka connectors work to move data from Kafka topics to external systems like JDBC databases, S3 storage, and Elasticsearch indexes. The flow starts with reading messages from Kafka, applying connector-specific configurations, transforming the data into the required format, and finally sending it to the sink. Each connector type (JDBC, S3, Elasticsearch) processes messages independently but follows the same steps. Variables like 'Message' and 'Data Sink' change as data moves through the steps. Key moments clarify why messages reset after sending and how connectors know their sink. The quiz tests understanding of outputs at different steps and connector behavior. This helps beginners see the step-by-step data movement in Kafka connectors.

Practice

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

easy

A. To monitor Kafka cluster health

B. To store Kafka data in cloud object storage

C. To enable full-text search on Kafka data

D. To move data between Kafka and relational databases

Solution

1. Step 1: Understand JDBC connector role

   JDBC connectors are designed to connect Kafka with relational databases using JDBC protocol.

2. Step 2: Match purpose with options

   To move data between Kafka and relational databases correctly states moving data between Kafka and databases, which is the JDBC connector's main use.

3. Final Answer:

   To move data between Kafka and relational databases -> Option D

4. Quick Check:

   JDBC connector = database integration [OK]

Hint: JDBC always relates to databases and Kafka [OK]

Common Mistakes:

• Confusing JDBC with S3 or Elasticsearch connectors
• Thinking JDBC stores data in cloud storage
• Assuming JDBC is for search functionality

2. Which of the following is the correct configuration property to specify the S3 bucket name in a Kafka S3 sink connector?

easy

A. s3.bucket.name

B. bucket.name.s3

C. s3.bucket

D. bucket.s3.name

Solution

1. Step 1: Recall S3 connector config syntax

   The standard property to specify the bucket is s3.bucket.name.

2. Step 2: Compare options with standard property

   Only s3.bucket.name matches the correct property name exactly.

3. Final Answer:

   s3.bucket.name -> Option A

4. Quick Check:

   Bucket name property = s3.bucket.name [OK]

Hint: S3 config keys start with 's3.' prefix [OK]

Common Mistakes:

• Swapping order of words in property name
• Using incorrect prefixes like 'bucket.s3'
• Missing the '.name' suffix

3. Given the following Kafka Elasticsearch sink connector config snippet:

{
  "connector.class": "io.confluent.connect.elasticsearch.ElasticsearchSinkConnector",
  "topics": "orders",
  "connection.url": "http://localhost:9200",
  "type.name": "_doc",
  "key.ignore": "false"
}

What will happen when a message with key 'order123' and value '{"item":"book"}' is sent to topic 'orders'?

medium

A. The message will be stored in Kafka only, not sent to Elasticsearch

B. The document will be indexed with a random ID ignoring the key

C. The document will be indexed in Elasticsearch with ID 'order123'

D. The connector will fail due to missing authentication

Solution

1. Step 1: Analyze key.ignore setting

   Since "key.ignore" is set to "false", the connector uses the Kafka message key as the Elasticsearch document ID.

2. Step 2: Understand indexing behavior

   The message with key 'order123' will be indexed in Elasticsearch with that key as the document ID.

3. Final Answer:

   The document will be indexed in Elasticsearch with ID 'order123' -> Option C

4. Quick Check:

   key.ignore=false means use key as ID [OK]

Hint: key.ignore=false means use Kafka key as ES document ID [OK]

Common Mistakes:

• Assuming key is ignored by default
• Thinking connector fails without auth (not always true)
• Confusing Kafka storage with Elasticsearch indexing

4. You configured a Kafka JDBC source connector but it is not importing any data. Which of the following is the most likely cause?

medium

A. The Kafka topic name is missing in the connector config

B. The JDBC URL is incorrect or unreachable

C. The connector class is set to ElasticsearchSinkConnector

D. The S3 bucket name is not specified

Solution

1. Step 1: Check connectivity issues

   If the JDBC URL is wrong or unreachable, the connector cannot read data from the database.

2. Step 2: Evaluate other options

   The Kafka topic name is missing in the connector config would cause errors but usually the connector won't start; The connector class is set to ElasticsearchSinkConnector is wrong connector type; The S3 bucket name is not specified is unrelated to JDBC source.

3. Final Answer:

   The JDBC URL is incorrect or unreachable -> Option B

4. Quick Check:

   Bad JDBC URL = no data import [OK]

Hint: Always verify JDBC URL connectivity first [OK]

Common Mistakes:

• Confusing connector class for source vs sink
• Ignoring missing topic config errors
• Checking unrelated S3 settings for JDBC issues

5. You want to archive Kafka topic data to S3 daily and also enable fast search on the same data using Elasticsearch. Which connector setup is the best approach?

hard

A. Use an S3 sink connector for archiving and an Elasticsearch sink connector for search, both consuming from the same Kafka topic

B. Use only an S3 sink connector and query S3 directly for search

C. Use a JDBC sink connector to store data in a database, then export to S3 and Elasticsearch from there

D. Use an Elasticsearch source connector to send data to S3

Solution

1. Step 1: Understand connector roles

   S3 sink connector archives data to cloud storage; Elasticsearch sink connector indexes data for search.

2. Step 2: Choose correct architecture

   Using both sink connectors on the same Kafka topic allows parallel archiving and search capabilities efficiently.

3. Final Answer:

   Use an S3 sink connector for archiving and an Elasticsearch sink connector for search, both consuming from the same Kafka topic -> Option A

4. Quick Check:

   Separate sinks for archive and search = best practice [OK]

Hint: Use separate sink connectors for storage and search [OK]

Common Mistakes:

• Trying to search data directly in S3 without indexing
• Using JDBC sink unnecessarily for this use case
• Confusing source and sink connector roles