Consumer poll loop in Kafka - Mini Project: Build & Apply

Consumer poll loop

📖 Scenario: You are building a simple Kafka consumer application that reads messages from a Kafka topic continuously. This is common in real-world systems where you want to process data streams as they arrive.

🎯 Goal: Build a Kafka consumer poll loop that continuously polls messages from a topic called test-topic and prints each message's key and value.

📋 What You'll Learn

Create a Kafka consumer with the group ID my-group.

Subscribe the consumer to the topic test-topic.

Set a poll timeout of 100 milliseconds.

Use a while True loop to continuously poll messages.

Print each message's key and value in the format: Key: {key}, Value: {value}.

💡 Why This Matters

🌍 Real World

Kafka consumers are used in real-time data processing systems to read streams of data such as logs, user activity, or sensor data.

💼 Career

Understanding how to write a consumer poll loop is essential for backend developers, data engineers, and anyone working with streaming data platforms.

1

Create Kafka consumer and subscribe to topic

Create a Kafka consumer called consumer with the group ID my-group and bootstrap server localhost:9092. Then subscribe the consumer to the topic test-topic.

Kafka

# Create Kafka consumer with group ID 'my-group' and subscribe to 'test-topic'
# Your code here

Need a hint?

Use KafkaConsumer from the kafka library. You can pass the topic name directly or use consumer.subscribe() after creating the consumer.

2

Set poll timeout variable

Create a variable called poll_timeout_ms and set it to 100 to represent the poll timeout in milliseconds.

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

# Set poll timeout to 100 milliseconds
# Your code here

Need a hint?

This variable will be used as the timeout value when polling messages.

3

Write the consumer poll loop

Write a while True loop that polls messages from consumer using poll_timeout_ms as the timeout. Iterate over the records returned by consumer.poll(timeout_ms=poll_timeout_ms) using nested loops (for partition_records in records.values(): for message in partition_records:) and print each message's key and value using print(f"Key: {message.key}, Value: {message.value}").

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

poll_timeout_ms = 100

# Write a while True loop to poll messages and print key and value
# Your code here

Need a hint?

Use nested loops because consumer.poll() returns a dictionary of partitions to lists of messages.

4

Print the consumer output

Run the program and observe the printed output lines showing message keys and values in the format Key: {key}, Value: {value}. For example, if a message has key b'user1' and value b'hello', the output should be Key: b'user1', Value: b'hello'.

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    group_id='my-group',
    bootstrap_servers=['localhost:9092']
)
consumer.subscribe(['test-topic'])

poll_timeout_ms = 100

while True:
    records = consumer.poll(timeout_ms=poll_timeout_ms)
    for partition_records in records.values():
        for message in partition_records:
            print(f"Key: {message.key}, Value: {message.value}")

# Run the program and observe output

Need a hint?

Make sure your Kafka broker is running and the topic test-topic has messages with keys and values.