Offset management in Kafka - Mini Project: Build & Apply

Offset management

📖 Scenario: You are building a simple Kafka consumer application that reads messages from a topic. To keep track of which messages have been processed, you need to manage offsets manually.

🎯 Goal: Build a Kafka consumer that reads messages from a topic, commits offsets manually after processing, and prints the committed offsets.

📋 What You'll Learn

Create a Kafka consumer subscribed to the topic 'test-topic'.

Configure the consumer to disable automatic offset commits.

Manually commit offsets after processing each message.

Print the committed offsets to verify manual offset management.

💡 Why This Matters

🌍 Real World

Kafka consumers often need to manage offsets manually to ensure messages are processed exactly once or to control when offsets are saved.

💼 Career

Understanding offset management is crucial for roles involving real-time data processing, streaming applications, and building reliable data pipelines.

1

Create Kafka consumer and subscribe to topic

Create a Kafka consumer called consumer with the following properties: bootstrap.servers='localhost:9092', group.id='test-group', enable.auto.commit=False, auto.offset.reset='earliest'. Then subscribe consumer to the topic 'test-topic'.

Kafka

# Create Kafka consumer with required properties and subscribe to 'test-topic'
# Your code here

Need a hint?

Use KafkaConsumer from the kafka library with the specified properties and subscribe to 'test-topic'.

2

Poll messages and process

Add a variable called message_count and set it to 0. Then poll messages from consumer using consumer.poll(timeout_ms=1000) and iterate over the records to increment message_count for each message.

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='localhost:9092',
    group_id='test-group',
    enable_auto_commit=False,
    auto_offset_reset='earliest'
)

# Create message_count variable and poll messages
# Your code here

Need a hint?

Use consumer.poll(timeout_ms=1000) to get messages and loop through them to count.

3

Manually commit offsets after processing

After processing messages, manually commit offsets using consumer.commit().

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='localhost:9092',
    group_id='test-group',
    enable_auto_commit=False,
    auto_offset_reset='earliest'
)

message_count = 0
records = consumer.poll(timeout_ms=1000)
for tp, messages in records.items():
    for message in messages:
        message_count += 1

# Commit offsets manually
# Your code here

Need a hint?

Call consumer.commit() after processing messages to save offsets manually.

4

Print committed offsets

Print the committed offsets using print(consumer.committed(tp)) inside the loop over records.items().

Kafka

from kafka import KafkaConsumer

consumer = KafkaConsumer(
    'test-topic',
    bootstrap_servers='localhost:9092',
    group_id='test-group',
    enable_auto_commit=False,
    auto_offset_reset='earliest'
)

message_count = 0
records = consumer.poll(timeout_ms=1000)
for tp, messages in records.items():
    for message in messages:
        message_count += 1
    # Print committed offset for this partition
    # Your code here

consumer.commit()

Need a hint?

Use print(f"Committed offset for {tp}: {consumer.committed(tp)}") inside the loop to show committed offsets.