Python consumer in Kafka - Step-by-Step Execution

Process Flow - Python consumer

Start Consumer

↓

Connect to Kafka Broker

↓

Subscribe to Topic

↓

Poll for Messages

↓

Message Received?

No→Wait and Poll Again

Yes↓

Process Message

↓

Commit Offset

↓

Repeat Polling Loop

↓

Stop Consumer on Signal

The Python consumer connects to Kafka, subscribes to a topic, polls messages in a loop, processes each message, commits offsets, and repeats until stopped.

Execution Sample

Kafka

from kafka import KafkaConsumer
consumer = KafkaConsumer('my_topic', bootstrap_servers='localhost:9092')
for message in consumer:
    print(f"Received: {message.value.decode()}")

This code connects to Kafka, subscribes to 'my_topic', and prints each message received.

Process Table

Step	Action	Evaluation	Result
1	Create KafkaConsumer instance	Connect to localhost:9092	Consumer ready, subscribed to 'my_topic'
2	Start polling loop	Wait for message	No message yet, keep polling
3	Message arrives	message.value = b'Hello'	Message received
4	Decode message	message.value.decode() = 'Hello'	Decoded message 'Hello'
5	Print message	print output	Output: Received: Hello
6	Commit offset	Offset committed	Ready for next message
7	Poll next message	Wait for message	No message yet, keep polling
8	Stop consumer	Signal received	Consumer stops gracefully

💡 Consumer stops when external signal triggers shutdown

Status Tracker

Variable	Start	After Step 3	After Step 4	After Step 6	Final
consumer	None	KafkaConsumer instance connected	Same instance	Same instance	Closed
message	None	b'Hello'	b'Hello'	b'Hello'	None
decoded_message	None	None	'Hello'	'Hello'	None

Key Moments - 3 Insights

Why does the consumer keep polling even when no messages arrive?

The consumer polls in a loop waiting for messages; if none arrive, it waits and polls again as shown in step 2 and 7 of the execution_table.

What does committing the offset mean and why is it important?

Committing the offset (step 6) tells Kafka the message was processed, so the consumer won't re-read it after restart, ensuring no duplicates.

Why do we decode message.value before printing?

Kafka messages are bytes; decoding converts bytes to readable text, as shown in step 4 where b'Hello' becomes 'Hello'.

Visual Quiz - 3 Questions

Look at the execution_table, what is the value of 'message' after step 3?

A b'Hello'

B 'Hello'

C None

D KafkaConsumer instance

💡 Hint

Check the 'message' variable in variable_tracker after step 3

At which step does the consumer commit the offset?

A Step 4

B Step 6

C Step 2

D Step 8

💡 Hint

Look for 'Commit offset' action in execution_table

If no messages arrive, what does the consumer do according to the execution_table?

A Stops immediately

B Prints 'Received: None'

C Keeps polling and waiting

D Throws an error

💡 Hint

See steps 2 and 7 where it waits and polls again

Concept Snapshot

Python Kafka Consumer:
- Create KafkaConsumer with topic and broker
- Poll messages in a loop
- Decode message bytes to string
- Process and commit offset
- Repeat until stopped
- Commit ensures no duplicate processing

Full Transcript

A Python Kafka consumer connects to the Kafka broker and subscribes to a topic. It enters a loop where it polls for messages. If no message is available, it waits and polls again. When a message arrives, it is received as bytes, decoded to a string, and processed (e.g., printed). After processing, the consumer commits the offset to mark the message as read. This loop continues until an external signal stops the consumer gracefully.

Practice

1.

What is the main purpose of a Python Kafka consumer?

easy

A. To create Kafka topics automatically

B. To send messages to Kafka topics

C. To read messages from Kafka topics and process them in Python

D. To monitor Kafka server health

Solution

1. Step 1: Understand Kafka consumer role

   A Kafka consumer reads data from Kafka topics to use in applications.

2. Step 2: Differentiate from producer

   Producers send messages, consumers read them. Monitoring and topic creation are separate tasks.

3. Final Answer:

   To read messages from Kafka topics and process them in Python -> Option C

4. Quick Check:

   Consumer = reads messages [OK]

Hint: Consumers read, producers write messages in Kafka [OK]

Common Mistakes:

• Confusing consumer with producer
• Thinking consumer creates topics
• Assuming consumer monitors server

2.

Which of the following is the correct way to create a KafkaConsumer in Python to read from topic my_topic on localhost?

from kafka import KafkaConsumer
consumer = KafkaConsumer(_____)

easy

A. my_topic, bootstrap_servers='localhost:9092'

B. "my_topic", bootstrap='localhost:9092'

C. "my_topic", servers='localhost:9092'

D. "my_topic", bootstrap_servers='localhost:9092'

Solution

1. Step 1: Check KafkaConsumer parameters

   KafkaConsumer expects topic name as string and bootstrap_servers keyword argument.

2. Step 2: Identify correct syntax

   "my_topic", bootstrap_servers='localhost:9092' uses correct string for topic and correct keyword bootstrap_servers with proper value.

3. Final Answer:

   "my_topic", bootstrap_servers='localhost:9092' -> Option D

4. Quick Check:

   Correct syntax uses bootstrap_servers keyword [OK]

Hint: Use bootstrap_servers='host:port' exactly [OK]

Common Mistakes:

• Omitting quotes around topic name
• Using wrong keyword like servers or bootstrap
• Passing topic name without quotes

3.

What will be the output of this code snippet if the Kafka topic test has one message with value b'hello'?

from kafka import KafkaConsumer
consumer = KafkaConsumer("test", bootstrap_servers='localhost:9092', auto_offset_reset='earliest', group_id='group1')
for msg in consumer:
    print(msg.value.decode('utf-8'))
    break

medium

A. hello

B. b'hello'

C. msg.value

D. Error: decode method not found

Solution

1. Step 1: Understand message value type

   Kafka messages are bytes, so msg.value is b'hello'.

2. Step 2: Decode bytes to string

   Calling decode('utf-8') converts b'hello' to 'hello' string, which is printed.

3. Final Answer:

   hello -> Option A

4. Quick Check:

   Decoded bytes print string [OK]

Hint: Decode bytes with decode('utf-8') to get string [OK]

Common Mistakes:

• Printing raw bytes without decoding
• Trying to decode string instead of bytes
• Expecting b'hello' printed as string

4.

Identify the error in this Kafka consumer code snippet:

from kafka import KafkaConsumer
consumer = KafkaConsumer('topic1', bootstrap_servers='localhost:9092')
for message in consumer:
    print(message.value.decode())
consumer.close()

medium

A. decode() requires an encoding argument

B. The consumer.close() is called outside the loop and never reached

C. Missing group_id parameter in KafkaConsumer

D. The for loop should use consumer.poll() instead

Solution

1. Step 1: Analyze loop and close call

   The for loop on consumer is infinite; code never reaches consumer.close() after loop.

2. Step 2: Understand consumer.close() usage

   To close consumer properly, loop must break or use context manager; otherwise close() is unreachable.

3. Final Answer:

   The consumer.close() is called outside the loop and never reached -> Option B

4. Quick Check:

   Infinite loop blocks close() call [OK]

Hint: Infinite loops prevent code after them from running [OK]

Common Mistakes:

• Assuming decode() needs encoding argument
• Thinking group_id is mandatory
• Confusing poll() with iteration

5.

You want to consume messages from two Kafka topics topicA and topicB using one Python KafkaConsumer instance. Which code snippet correctly sets this up?

from kafka import KafkaConsumer
consumer = KafkaConsumer(_____)
for msg in consumer:
    print(f"Topic: {msg.topic}, Message: {msg.value.decode()}")

hard

A. ["topicA", "topicB"], bootstrap_servers='localhost:9092', auto_offset_reset='earliest', group_id='group1'

B. "topicA,topicB", bootstrap_servers='localhost:9092', group_id='group1'

C. "topicA" and "topicB", bootstrap_servers='localhost:9092'

D. "topicA"; "topicB", bootstrap_servers='localhost:9092', group_id='group1'

Solution

1. Step 1: Check how to subscribe to multiple topics

   KafkaConsumer accepts a list of topic names to subscribe to multiple topics.

2. Step 2: Verify correct syntax for multiple topics

   ["topicA", "topicB"], bootstrap_servers='localhost:9092', auto_offset_reset='earliest', group_id='group1' correctly passes a list ["topicA", "topicB"] and required parameters.

3. Final Answer:

   ["topicA", "topicB"], bootstrap_servers='localhost:9092', auto_offset_reset='earliest', group_id='group1' -> Option A

4. Quick Check:

   List of topics subscribes to multiple topics [OK]

Hint: Use list of topic names to consume multiple topics [OK]

Common Mistakes:

• Passing topics as comma-separated string
• Using logical operators like and or ; between topics
• Omitting group_id for consumer group