Python consumer in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer code snippet?

Consider the following Python Kafka consumer code using confluent_kafka. What will be printed?

Kafka

from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'testgroup', 'auto.offset.reset': 'earliest'}
consumer = Consumer(conf)
consumer.subscribe(['mytopic'])

msg = consumer.poll(timeout=1.0)
if msg is None:
    print('No message received')
elif msg.error():
    print(f'Error: {msg.error()}')
else:
    print(f'Received message: {msg.value().decode("utf-8")}')
consumer.close()

A Error: KafkaError{code=_UNKNOWN_TOPIC_OR_PART}

B Received message: <message content>

C No message received

D Raises a TypeError

💡 Hint

If the topic has no messages or Kafka is not running, poll returns None.

The poll method waits for a message. If none arrives within 1 second, it returns None. So the code prints 'No message received'.

❓ Predict Output

intermediate

What does this Kafka consumer print when a message with value 'hello' is received?

Given this code snippet, what will be printed if a message with value b'hello' is received?

Kafka

from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'grp1', 'auto.offset.reset': 'earliest'}
consumer = Consumer(conf)
consumer.subscribe(['topic1'])

msg = consumer.poll(timeout=2.0)
if msg and not msg.error():
    print(f'Message: {msg.value().decode()}')
else:
    print('No valid message')
consumer.close()

A Message: b'hello'

B Raises a UnicodeDecodeError

C No valid message

D Message: hello

💡 Hint

Message values are bytes and need decoding to strings.

The message value is bytes b'hello'. Decoding it with decode() converts it to string 'hello'.

🧠 Conceptual

advanced

Which error will this Kafka consumer code raise?

What error will this code raise when run?

Kafka

from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'grp2'}
consumer = Consumer(conf)
consumer.subscribe(['topic2'])

msg = consumer.poll(timeout=1.0)
print(msg.value().decode('utf-8'))
consumer.close()

A No error, prints message value

B AttributeError: 'NoneType' object has no attribute 'value'

C UnicodeDecodeError

D KafkaException

💡 Hint

If poll returns None, calling value() causes an error.

If no message is received, poll returns None. Calling value() on None raises AttributeError.

❓ Predict Output

advanced

What is the output of this Kafka consumer loop?

What will this code print if the topic has three messages with values 'one', 'two', 'three'?

Kafka

from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'grp3', 'auto.offset.reset': 'earliest'}
consumer = Consumer(conf)
consumer.subscribe(['topic3'])

count = 0
while count < 3:
    msg = consumer.poll(timeout=1.0)
    if msg and not msg.error():
        print(msg.value().decode())
        count += 1
consumer.close()

A one\ntwo\nthree

B three\ntwo\none

C No output

D Raises a RuntimeError

💡 Hint

The consumer reads messages in order from the topic.

The consumer polls and prints each message in the order they were produced: 'one', 'two', 'three'.

🔧 Debug

expert

Why does this Kafka consumer code hang indefinitely?

Given this code, why does it never exit the loop?

Kafka

from confluent_kafka import Consumer

conf = {'bootstrap.servers': 'localhost:9092', 'group.id': 'grp4', 'auto.offset.reset': 'latest'}
consumer = Consumer(conf)
consumer.subscribe(['topic4'])

while True:
    msg = consumer.poll(timeout=0.5)
    if msg is None:
        continue
    if msg.error():
        print(f'Error: {msg.error()}')
        break
    print(msg.value().decode())
consumer.close()

A Because 'auto.offset.reset' is 'latest' and no new messages arrive, poll returns None forever

B Because the loop condition is always True without a break on valid messages

C Because consumer.close() is never called inside the loop

D Because poll timeout is too short causing infinite retries

💡 Hint

Think about what happens if the consumer starts after messages were produced and 'auto.offset.reset' is 'latest'.

With 'auto.offset.reset' set to 'latest', the consumer waits for new messages after the current end. If no new messages arrive, poll returns None repeatedly, causing the loop to continue forever.

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