Consumer API basics in Kafka - Step-by-Step Execution

Process Flow - Consumer API basics

Start Consumer

↓

Subscribe to Topic

↓

Poll for Messages

↓

Process Messages

↓

Commit Offsets

↓

Repeat Polling

↩Back to Poll for Messages

The consumer starts, subscribes to a topic, polls messages, processes them, commits offsets, and repeats until stopped.

Execution Sample

Kafka

consumer = KafkaConsumer('my_topic')
consumer.subscribe(['my_topic'])
for message in consumer:
    print(message.value)
    consumer.commit()

This code creates a consumer, subscribes to 'my_topic', reads messages one by one, prints their values, and commits offsets.

Process Table

Step	Action	Message Polled	Offset	Output	Commit Offset
1	Start Consumer and Subscribe	None	None	None	None
2	Poll for message	Message1	0	Print Message1	Commit offset 1
3	Poll for message	Message2	1	Print Message2	Commit offset 2
4	Poll for message	Message3	2	Print Message3	Commit offset 3
5	Poll for message	No new message	2	No output	No commit
6	Stop Consumer	None	2	None	None

💡 No new messages available, consumer stops polling.

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 4	After Step 5	Final
message	None	Message1	Message2	Message3	None	None
offset	None	0	1	2	2	2
committed_offset	None	1	2	3	3	3

Key Moments - 3 Insights

Why do we commit offsets after processing each message?

Committing offsets tells Kafka which messages have been processed, so if the consumer restarts, it continues from the last committed offset (see steps 2-4 in execution_table).

What happens if no new messages are available during polling?

The poll returns no messages, so the consumer waits or stops polling (see step 5 in execution_table where no new message is polled).

Why do we subscribe to a topic before polling?

Subscription tells the consumer which topic(s) to listen to, so polling knows where to get messages from (see step 1 in execution_table).

Visual Quiz - 3 Questions

Look at the execution_table, what is the committed_offset after step 3?

A 2

B 1

C 0

D None

💡 Hint

Check the committed_offset value in variable_tracker after step 3.

At which step does the consumer find no new messages?

A Step 2

B Step 5

C Step 3

D Step 6

💡 Hint

Look at the 'Message Polled' column in execution_table.

If we skip committing offsets, what would happen on restart?

A Consumer starts from next offset after last commit

B Consumer starts from last committed offset

C Consumer starts from beginning of topic

D Consumer cannot start

💡 Hint

Refer to key_moments about why committing offsets is important.

Concept Snapshot

Kafka Consumer API basics:
- Create consumer and subscribe to topic
- Poll messages in a loop
- Process each message
- Commit offsets after processing
- Repeat polling until stopped
- Committing offsets ensures no message loss or duplication

Full Transcript

This visual execution shows how a Kafka consumer works. First, the consumer starts and subscribes to a topic. Then it polls messages one by one. Each message is processed by printing its value. After processing, the consumer commits the offset to mark the message as done. This repeats until no new messages are available, then the consumer stops. Committing offsets is important to avoid reprocessing messages after restart. If no messages are found during polling, the consumer waits or stops polling. Subscribing to the topic tells the consumer where to get messages from.

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