Python consumer in Kafka - Interactive Code Practice

Practice - 5 Tasks

1fill in blank

easy

Complete the code to import the KafkaConsumer class from the kafka module.

Kafka

from kafka import [1]

consumer = KafkaConsumer('my_topic')

Drag options to blanks, or click blank then click option'

AProducer

BKafkaConsumer

CKafkaProducer

DConsumerGroup

Attempts:

3 left

💡 Hint

Common Mistakes

Importing KafkaProducer instead of KafkaConsumer

Using Producer instead of KafkaConsumer

You need to import KafkaConsumer to create a consumer that reads messages from Kafka.

2fill in blank

medium

Complete the code to create a KafkaConsumer that reads from the topic 'orders'.

Kafka

consumer = KafkaConsumer([1])

Drag options to blanks, or click blank then click option'

A'orders'

B'payments'

C'users'

D'logs'

Attempts:

3 left

💡 Hint

Common Mistakes

Using the wrong topic name

Forgetting to put quotes around the topic name

The consumer should subscribe to the topic named 'orders' to read order messages.

3fill in blank

hard

Fix the error in the code to correctly consume messages from Kafka.

Kafka

for message in consumer.[1]():
    print(message.value)

Drag options to blanks, or click blank then click option'

Apoll

Bconsume

Csend

Dsubscribe

Attempts:

3 left

💡 Hint

Common Mistakes

Using 'send' which is for producers

Using 'subscribe' which subscribes to topics but does not fetch messages

The method to iterate over messages is poll() or simply iterating over the consumer. Here, poll() is the correct method to get messages.

4fill in blank

hard

Fill both blanks to create a consumer that reads from 'events' topic and uses 'localhost:9092' as the Kafka server.

Kafka

consumer = KafkaConsumer([1], bootstrap_servers=[2])

Drag options to blanks, or click blank then click option'

A'events'

B'localhost:9092'

C'127.0.0.1:9092'

D'my_topic'

Attempts:

3 left

💡 Hint

Common Mistakes

Using wrong topic name

Using wrong server address format

The topic name is 'events' and the Kafka server address is 'localhost:9092'.

5fill in blank

hard

Fill all three blanks to create a KafkaConsumer that reads from 'logs', connects to 'broker:9092', and uses 'earliest' offset reset.

Kafka

consumer = KafkaConsumer([1], bootstrap_servers=[2], auto_offset_reset=[3])

Drag options to blanks, or click blank then click option'

A'logs'

B'broker:9092'

C'earliest'

D'latest'

Attempts:

3 left

💡 Hint

Common Mistakes

Using 'latest' instead of 'earliest' for offset reset

Wrong server address format

Wrong topic name

The topic is 'logs', the server is 'broker:9092', and the offset reset policy is 'earliest' to start from the beginning of the topic.

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