Consumer API basics in Kafka - Mini Project: Build & Apply

Consumer API basics

📖 Scenario: You are working with Apache Kafka to read messages from a topic. Kafka is like a messaging system where producers send messages and consumers read them. In this project, you will create a simple Kafka consumer that reads messages from a topic called test-topic.

🎯 Goal: Build a Kafka consumer that connects to a Kafka server, subscribes to the test-topic, reads messages, and prints them to the console.

📋 What You'll Learn

Create a Kafka consumer configuration with the correct server and group ID

Subscribe the consumer to the test-topic

Poll messages from the topic

Print the received messages to the console

💡 Why This Matters

🌍 Real World

Kafka consumers are used in real systems to process streams of data like logs, user activity, or sensor data in real time.

💼 Career

Understanding Kafka consumer basics is essential for roles in data engineering, backend development, and DevOps where streaming data processing is common.

1

Create Kafka consumer configuration

Create a dictionary called consumer_config with these exact entries: 'bootstrap.servers': 'localhost:9092', 'group.id': 'my-group', 'auto.offset.reset': 'earliest'.

Kafka

# Create the consumer_config dictionary with bootstrap.servers, group.id, and auto.offset.reset
# Your code here

Need a hint?

Use a Python dictionary with the exact keys and values as shown.

2

Create Kafka consumer instance

Import Consumer from confluent_kafka and create a variable called consumer by passing consumer_config to Consumer().

Kafka

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

# Import Consumer and create consumer instance
# Your code here

Need a hint?

Use from confluent_kafka import Consumer and then create consumer = Consumer(consumer_config).

3

Subscribe to topic and poll messages

Use consumer.subscribe(['test-topic']) to subscribe to the topic. Then write a for loop that runs 5 times, inside the loop call consumer.poll(1.0) to get a message and store it in msg. Check if msg is not null and not an error, then print msg.value().decode('utf-8').

Kafka

from confluent_kafka import Consumer

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(consumer_config)

# Subscribe to 'test-topic' and poll 5 messages
# Your code here

Need a hint?

Subscribe with consumer.subscribe(['test-topic']). Use a for loop to poll 5 messages. Check for msg not null and no error before printing.

4

Close the consumer and print completion message

After the loop, call consumer.close() to close the consumer. Then write print("Finished reading messages") to display the completion message.

Kafka

from confluent_kafka import Consumer

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'my-group',
    'auto.offset.reset': 'earliest'
}

consumer = Consumer(consumer_config)

consumer.subscribe(['test-topic'])

for _ in range(5):
    msg = consumer.poll(1.0)
    if msg is not null and not msg.error():
        print(msg.value().decode('utf-8'))

# Close the consumer and print "Finished reading messages"
# Your code here

Need a hint?

Call consumer.close() after the loop. Then print the exact message "Finished reading messages".