Consumer API basics in Kafka - Mini Project: Build & Apply

Consumer API basics

📖 Scenario: You are working with Apache Kafka to read messages from a topic. Kafka is like a messaging system where producers send messages and consumers read them. In this project, you will create a simple Kafka consumer that reads messages from a topic called test-topic.

🎯 Goal: Build a Kafka consumer that connects to a Kafka server, subscribes to the test-topic, reads messages, and prints them to the console.

📋 What You'll Learn

Create a Kafka consumer configuration with the correct server and group ID

Subscribe the consumer to the test-topic

Poll messages from the topic

Print the received messages to the console

💡 Why This Matters

🌍 Real World

Kafka consumers are used in real systems to process streams of data like logs, user activity, or sensor data in real time.

💼 Career

Understanding Kafka consumer basics is essential for roles in data engineering, backend development, and DevOps where streaming data processing is common.

1

Create Kafka consumer configuration

Create a dictionary called consumer_config with these exact entries: 'bootstrap.servers': 'localhost:9092', 'group.id': 'my-group', 'auto.offset.reset': 'earliest'.

Kafka

# Create the consumer_config dictionary with bootstrap.servers, group.id, and auto.offset.reset
# Your code here

Hint

Use a Python dictionary with the exact keys and values as shown.

2

Create Kafka consumer instance

Import Consumer from confluent_kafka and create a variable called consumer by passing consumer_config to Consumer().

Kafka

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

# Import Consumer and create consumer instance
# Your code here

Hint

Use from confluent_kafka import Consumer and then create consumer = Consumer(consumer_config).

3

Subscribe to topic and poll messages

Use consumer.subscribe(['test-topic']) to subscribe to the topic. Then write a for loop that runs 5 times, inside the loop call consumer.poll(1.0) to get a message and store it in msg. Check if msg is not null and not an error, then print msg.value().decode('utf-8').

Kafka

from confluent_kafka import Consumer

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(consumer_config)

# Subscribe to 'test-topic' and poll 5 messages
# Your code here

Hint

Subscribe with consumer.subscribe(['test-topic']). Use a for loop to poll 5 messages. Check for msg not null and no error before printing.

4

Close the consumer and print completion message

After the loop, call consumer.close() to close the consumer. Then write print("Finished reading messages") to display the completion message.

Kafka

from confluent_kafka import Consumer

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(consumer_config)

consumer.subscribe(['test-topic'])

for _ in range(5):
    msg = consumer.poll(1.0)
    if msg is not null and not msg.error():
        print(msg.value().decode('utf-8'))

# Close the consumer and print "Finished reading messages"
# Your code here

Hint

Call consumer.close() after the loop. Then print the exact message "Finished reading messages".

Practice

1. What is the primary purpose of the Kafka Consumer API?

easy

A. To send messages to Kafka topics

B. To delete Kafka topics

C. To create Kafka topics

D. To read messages from Kafka topics

Solution

1. Step 1: Understand Kafka Consumer API role

   The Consumer API is designed to read or consume messages from Kafka topics.

2. Step 2: Differentiate from Producer API

   The Producer API sends messages, while Consumer API reads them.

3. Final Answer:

   To read messages from Kafka topics -> Option D

4. Quick Check:

   Consumer API = Read messages [OK]

Hint: Consumer API reads messages, Producer API sends messages [OK]

Common Mistakes:

• Confusing Consumer API with Producer API
• Thinking Consumer API creates or deletes topics
• Mixing up reading and writing operations

2. Which of the following is the correct way to create a Kafka consumer in Java?

easy

A. new KafkaProducer<>(properties);

B. new KafkaConsumer<>(properties);

C. KafkaConsumer.create(properties);

D. KafkaConsumer.start(properties);

Solution

1. Step 1: Recall Kafka consumer instantiation syntax

   Kafka consumer is created by calling the constructor: new KafkaConsumer<>(properties);

2. Step 2: Eliminate incorrect options

   Producer is for sending messages, so new KafkaProducer<>(properties); is wrong. Options C and D are not valid KafkaConsumer methods.

3. Final Answer:

   new KafkaConsumer<>(properties); -> Option B

4. Quick Check:

   Constructor new KafkaConsumer<>(props) = Correct [OK]

Hint: Use new KafkaConsumer<>(props) to create consumer [OK]

Common Mistakes:

• Using KafkaProducer instead of KafkaConsumer
• Calling non-existent static methods
• Confusing method names for creation

3. Given the following Java code snippet, what will be the output if the topic has two messages "Hello" and "World"?

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "test-group");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(Collections.singletonList("my-topic"));
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
    System.out.println(record.value());
}

medium

A. Compilation error

B. No output

C. Hello\nWorld

D. Runtime exception

Solution

1. Step 1: Understand consumer subscription and poll

   The consumer subscribes to "my-topic" and polls messages for 1 second.

2. Step 2: Process messages in the topic

   Since the topic has two messages "Hello" and "World", the poll will fetch both, and the loop prints each value.

3. Final Answer:

   Hello\nWorld -> Option C

4. Quick Check:

   poll() fetches messages = Hello and World printed [OK]

Hint: poll() fetches messages; loop prints each value [OK]

Common Mistakes:

• Assuming poll returns no messages immediately
• Confusing subscribe with assign
• Expecting compile or runtime errors without cause

4. Identify the error in this Kafka consumer code snippet:

Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
props.put("group.id", "group1");
props.put("key.deserializer", "StringDeserializer");
props.put("value.deserializer", "StringDeserializer");
KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe(List.of("topic1"));
ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
for (ConsumerRecord record : records) {
    System.out.println(record.value());
}

medium

A. Deserializer class names are incomplete and will cause ClassNotFoundException

B. Missing call to consumer.close()

C. Incorrect subscription method used

D. poll() duration is too short and causes timeout

Solution

1. Step 1: Check deserializer configuration

   The deserializer class names must be fully qualified, e.g., "org.apache.kafka.common.serialization.StringDeserializer".

2. Step 2: Identify impact of incorrect class names

   Using "StringDeserializer" alone will cause ClassNotFoundException at runtime.

3. Final Answer:

   Deserializer class names are incomplete and will cause ClassNotFoundException -> Option A

4. Quick Check:

   Deserializer full class name required = Deserializer class names are incomplete and will cause ClassNotFoundException [OK]

Hint: Use full deserializer class names to avoid errors [OK]

Common Mistakes:

• Using short class names for deserializers
• Ignoring missing consumer.close() (not error here)
• Assuming poll duration causes errors

5. You want to process messages from multiple topics and commit offsets manually after processing. Which code snippet correctly sets up the Kafka consumer for this use case?

hard

A. props.put("enable.auto.commit", "false");
consumer.subscribe(Arrays.asList("topicA", "topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
consumer.commitSync();
}

B. props.put("enable.auto.commit", "true");
consumer.subscribe(Collections.singletonList("topicA"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
}

C. props.put("enable.auto.commit", "false");
consumer.assign(Arrays.asList(new TopicPartition("topicA", 0)));
consumer.subscribe(Arrays.asList("topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
consumer.commitAsync();
}

D. props.put("enable.auto.commit", "false");
consumer.subscribe(Arrays.asList("topicA", "topicB"));
while (true) {
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
process(record.value());
}
}

Solution

1. Step 1: Disable auto commit for manual offset control

   Setting "enable.auto.commit" to "false" allows manual offset commits.

2. Step 2: Subscribe to multiple topics correctly

   Use consumer.subscribe(Arrays.asList("topicA", "topicB")) to listen to multiple topics.

3. Step 3: Commit offsets after processing

   Calling consumer.commitSync() after processing ensures offsets are saved manually.

4. Final Answer:

   Code snippet with manual commit, multiple topic subscription, and commitSync() -> Option A

5. Quick Check:

   Manual commit + multiple topics = props.put("enable.auto.commit", "false"); consumer.subscribe(Arrays.asList("topicA", "topicB")); while (true) { ConsumerRecords records = consumer.poll(Duration.ofMillis(1000)); for (ConsumerRecord record : records) { process(record.value()); } consumer.commitSync(); } [OK]

Hint: Disable auto commit and call commitSync() after processing [OK]

Common Mistakes:

• Using auto commit when manual commit is needed
• Mixing assign() and subscribe() incorrectly
• Forgetting to commit offsets after processing