Why consumers process messages in Kafka - Visual Breakdown

Process Flow - Why consumers process messages

Message produced to topic

↓

Message stored in Kafka partition

↓

Consumer polls messages

↓

Consumer processes message

↓

Consumer commits offset

↓

Ready for next message

This flow shows how a message is produced, stored, consumed, processed, and acknowledged by committing the offset.

Execution Sample

Kafka

consumer = KafkaConsumer('topic', bootstrap_servers='localhost:9092', enable_auto_commit=False)
for message in consumer:
    process(message)
    consumer.commit()

This code shows a consumer reading messages from a Kafka topic, processing each message, and committing the offset.

Process Table

Step	Action	Message Offset	Processing Status	Offset Commit Status
1	Poll message at offset 0	0	Not processed	Not committed
2	Process message at offset 0	0	Processed	Not committed
3	Commit offset 0	0	Processed	Committed
4	Poll message at offset 1	1	Not processed	Committed offset 0
5	Process message at offset 1	1	Processed	Committed offset 0
6	Commit offset 1	1	Processed	Committed
7	No more messages	-	-	-

💡 No more messages to consume, consumer waits for new messages.

Status Tracker

Variable	Start	After Step 2	After Step 3	After Step 5	After Step 6	Final
message_offset	-	0	0	1	1	1
processing_status	Not processed	Processed	Processed	Processed	Processed	Processed
offset_commit_status	Not committed	Not committed	Committed	Committed offset 0	Committed	Committed

Key Moments - 3 Insights

Why does the consumer commit the offset after processing the message?

Committing the offset tells Kafka that the message was processed successfully, so the consumer won't reprocess it. See execution_table steps 3 and 6.

What happens if the consumer processes a message but does not commit the offset?

If the offset is not committed, Kafka will resend the message on restart, causing duplicate processing. This is why committing is important (execution_table step 2 vs 3).

Why does the consumer poll messages one by one in the flow?

Polling fetches messages from Kafka partitions in order. Processing them one by one ensures correct sequence and offset tracking (see execution_table steps 1,4).

Visual Quiz - 3 Questions

Look at the execution_table, what is the processing status at step 2?

A Committed

B Not processed

C Processed

D Polling

💡 Hint

Check the 'Processing Status' column at step 2 in the execution_table.

At which step does the consumer commit the offset for the first message?

A Step 3

B Step 1

C Step 2

D Step 4

💡 Hint

Look at the 'Offset Commit Status' column to find when it changes to 'Committed'.

If the consumer skips committing the offset, what will happen on restart?

A Messages will be lost

B Messages will be reprocessed

C Consumer will crash

D No effect

💡 Hint

Refer to key_moments about offset commit importance and execution_table steps 2 and 3.

Concept Snapshot

Kafka consumers process messages by polling from partitions.
Each message is processed in order.
After processing, the consumer commits the offset.
Committing offset prevents reprocessing on restart.
This ensures reliable and ordered message handling.

Full Transcript

In Kafka, messages are produced to topics and stored in partitions. Consumers poll these messages one by one. Each message has an offset number. The consumer processes the message and then commits the offset to Kafka. Committing the offset tells Kafka that the message was handled successfully. This prevents the consumer from processing the same message again if it restarts. The flow is: poll message, process it, commit offset, then move to the next message. This ensures messages are processed reliably and in order.

Practice

1. Why do Kafka consumers process messages from topics?

easy

A. To create new topics automatically

B. To read and use data sent by producers

C. To delete messages from the topic

D. To send messages to other consumers

Solution

1. Step 1: Understand the role of consumers in Kafka

   Consumers read messages produced by producers to get data for processing.

2. Step 2: Identify the purpose of processing messages

   Processing messages allows applications to react to new information sent by producers.

3. Final Answer:

   To read and use data sent by producers -> Option B

4. Quick Check:

   Consumers read messages = To use data [OK]

Hint: Consumers read messages to get data from producers [OK]

Common Mistakes:

• Thinking consumers send messages
• Confusing consumers with producers
• Believing consumers delete messages

2. Which of the following is the correct way for a Kafka consumer to subscribe to a topic named orders in Java?

easy

A. consumer.listen("orders");

B. consumer.subscribe("orders");

C. consumer.connect("orders");

D. consumer.subscribe(Collections.singletonList("orders"));

Solution

1. Step 1: Recall Kafka consumer subscription syntax in Java

   The consumer.subscribe method expects a collection of topic names, usually a list.

2. Step 2: Check each option's syntax

   consumer.subscribe(Collections.singletonList("orders")); uses Collections.singletonList("orders"), which is correct. Options A, B, and D use incorrect methods or argument types.

3. Final Answer:

   consumer.subscribe(Collections.singletonList("orders")); -> Option D

4. Quick Check:

   subscribe needs a list of topics [OK]

Hint: Use subscribe with a list of topics in Java [OK]

Common Mistakes:

• Passing a string instead of a list
• Using non-existent methods like listen or connect
• Forgetting to import Collections

3. Given the following Kafka consumer code snippet in Python, what will be printed?

from kafka import KafkaConsumer

consumer = KafkaConsumer('test-topic', bootstrap_servers=['localhost:9092'])
for message in consumer:
    print(message.value.decode('utf-8'))
    break

medium

A. The first message's value from 'test-topic' decoded as UTF-8

B. All messages from 'test-topic' printed continuously

C. An error because decode is not a method

D. No output because the loop breaks immediately

Solution

1. Step 1: Understand the consumer loop behavior

   The for loop reads messages from 'test-topic' one by one.

2. Step 2: Analyze the print and break statements

   The first message's value is decoded from bytes to string and printed, then the loop breaks immediately.

3. Final Answer:

   The first message's value from 'test-topic' decoded as UTF-8 -> Option A

4. Quick Check:

   Print first message then break = first message only [OK]

Hint: Loop breaks after first message, so only one prints [OK]

Common Mistakes:

• Thinking all messages print
• Confusing decode method usage
• Assuming no output due to break

4. What is wrong with this Kafka consumer code snippet in Java?

KafkaConsumer consumer = new KafkaConsumer<>(props);
consumer.subscribe("my-topic");
while(true) {
    ConsumerRecords records = consumer.poll(Duration.ofMillis(100));
    for (ConsumerRecord record : records) {
        System.out.println(record.value());
    }
}

medium

A. poll method should not have a timeout argument

B. ConsumerRecord does not have a value() method

C. subscribe method requires a collection, not a single string

D. The while loop should be replaced with a for loop

Solution

1. Step 1: Check the subscribe method usage

   In Kafka Java client, subscribe expects a collection of topic names, not a single string.

2. Step 2: Verify other code parts

   poll with Duration is correct, ConsumerRecord has value(), and while(true) is valid for continuous consumption.

3. Final Answer:

   subscribe method requires a collection, not a single string -> Option C

4. Quick Check:

   subscribe needs list of topics [OK]

Hint: subscribe needs a list, not a single string [OK]

Common Mistakes:

• Passing a string directly to subscribe
• Misunderstanding poll method arguments
• Thinking ConsumerRecord lacks value()

5. You want your Kafka consumer to process messages only from the latest offset to react quickly to new data. Which consumer configuration should you set before subscribing?

hard

A. "auto.offset.reset" to "latest"

B. "enable.auto.commit" to false

C. "group.id" to a unique value each time

D. "max.poll.records" to 1

Solution

1. Step 1: Understand offset reset behavior

   Setting "auto.offset.reset" to "latest" makes the consumer start reading from the newest messages if no committed offset exists.

2. Step 2: Evaluate other options

   "enable.auto.commit" controls offset commits, "group.id" identifies consumer groups, and "max.poll.records" limits batch size but doesn't affect starting offset.

3. Final Answer:

   "auto.offset.reset" to "latest" -> Option A

4. Quick Check:

   Start from latest messages = auto.offset.reset=latest [OK]

Hint: Set auto.offset.reset to latest to read newest messages [OK]

Common Mistakes:

• Confusing offset reset with auto commit
• Changing group.id to control offset start
• Using max.poll.records to control offset