Common connectors (JDBC, S3, Elasticsearch) in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

Kafka Connect JDBC Source Connector Configuration Output

What will be the output when the following Kafka Connect JDBC Source connector configuration is used to import data from a MySQL database table named users?

Kafka

connector.class=io.confluent.connect.jdbc.JdbcSourceConnector
connection.url=jdbc:mysql://localhost:3306/mydb
connection.user=root
connection.password=password
mode=incrementing
incrementing.column.name=id
topic.prefix=mysql-
table.whitelist=users

A Kafka topic 'mysql-users' will receive data but only for columns excluding 'id'

B Kafka topic 'users' will receive all rows from 'users' table in bulk mode

C Kafka topic 'mysql-users' will receive only the latest snapshot of 'users' table without incremental updates

D Kafka topic 'mysql-users' will receive all rows from 'users' table incrementally based on 'id' column

💡 Hint

Look at the mode and incrementing.column.name properties to understand how data is imported.

The configuration uses 'incrementing' mode with 'id' column, so Kafka Connect imports new rows incrementally into topic 'mysql-users'.

❓ Predict Output

intermediate

Kafka Connect S3 Sink Connector Behavior

Given the following Kafka Connect S3 Sink connector configuration, what will be the result in the S3 bucket after running the connector?

Kafka

connector.class=io.confluent.connect.s3.S3SinkConnector
s3.bucket.name=my-kafka-bucket
topics=my-topic
flush.size=3
storage.class=io.confluent.connect.s3.storage.S3Storage
format.class=io.confluent.connect.s3.format.json.JsonFormat

A Data from 'my-topic' will be written to S3 in CSV format with 3 records per file

B Data from 'my-topic' will be written to S3 as a single large JSON file with all records

C Data from 'my-topic' will be written to S3 in JSON files, each containing 3 records

D Data from 'my-topic' will not be written to S3 because flush.size is too small

💡 Hint

Check the flush.size and format.class properties to understand file creation.

The connector writes data in JSON format to S3, creating a new file every 3 records as specified by flush.size.

❓ Predict Output

advanced

Kafka Connect Elasticsearch Sink Connector Document ID Behavior

What will be the document ID in Elasticsearch for the following Kafka Connect Elasticsearch Sink connector configuration when writing records from topic 'orders'?

Kafka

connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
connection.url=http://localhost:9200
topics=orders
key.ignore=false
schema.ignore=true
behavior.on.null.values=delete

A The document ID will be the Kafka record key from topic 'orders'

B The document ID will be auto-generated by Elasticsearch ignoring Kafka keys

C The connector will fail because key.ignore is false but keys are missing

D The document ID will be the Kafka record value converted to string

💡 Hint

key.ignore=false means keys are used as document IDs.

With key.ignore=false, the connector uses Kafka record keys as Elasticsearch document IDs, ensuring updates overwrite existing documents.

🧠 Conceptual

advanced

Error Handling in Kafka Connect S3 Sink Connector

Which of the following best describes how Kafka Connect S3 Sink connector handles errors when writing data to S3?

A It retries failed writes a configurable number of times and then fails the task if unsuccessful

B It drops failed records silently and continues processing without alerting

C It retries failed writes indefinitely until success, blocking all other processing

D It writes failed records to a local file and skips them in S3

💡 Hint

Consider how Kafka Connect tasks handle transient errors and retries.

Kafka Connect retries failed writes based on configuration and fails the task if retries are exhausted, allowing for error visibility and recovery.

❓ Predict Output

expert

Kafka Connect Elasticsearch Sink Connector Bulk Indexing Behavior

Given this Kafka Connect Elasticsearch Sink connector configuration, what is the expected behavior regarding bulk indexing and document updates?

Kafka

connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
connection.url=http://localhost:9200
topics=products
batch.size=200
max.in.flight.requests=5
key.ignore=false
behavior.on.null.values=ignore

A The connector sends one record at a time to Elasticsearch ignoring keys, creating new documents only

B The connector sends batches of up to 200 records concurrently (up to 5 batches) using Kafka keys as document IDs, updating existing documents

C The connector sends batches of 200 records sequentially, ignoring keys and overwriting documents randomly

D The connector fails because max.in.flight.requests cannot be greater than 1 with batch.size set

💡 Hint

Check batch.size, max.in.flight.requests, and key.ignore settings for concurrency and document ID usage.

The connector batches records up to batch.size and sends multiple batches concurrently up to max.in.flight.requests, using keys as document IDs to update documents.

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