Consumer offset commit strategies in Kafka - Mini Project: Build & Apply

Consumer offset commit strategies

📖 Scenario: You are building a Kafka consumer application that reads messages from a topic. To ensure your application processes messages reliably, you need to manage how and when the consumer commits the offsets of messages it has processed.This project will guide you through setting up a Kafka consumer, configuring offset commit strategies, and printing the committed offsets.

🎯 Goal: Build a Kafka consumer that reads messages from a topic, uses a specific offset commit strategy, and prints the committed offsets to verify the strategy works.

📋 What You'll Learn

Create a Kafka consumer configuration with a specific group id and offset reset policy

Add a configuration variable to control the offset commit strategy (automatic or manual)

Implement the consumer logic to commit offsets according to the chosen strategy

Print the committed offsets after processing messages

💡 Why This Matters

🌍 Real World

Kafka consumers must manage offset commits to avoid processing messages multiple times or missing messages. Choosing the right commit strategy helps build reliable data pipelines.

💼 Career

Understanding consumer offset commit strategies is essential for roles involving real-time data processing, streaming applications, and building fault-tolerant systems using Kafka.

1

Create Kafka consumer configuration

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

Kafka

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

Hint

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

2

Add offset commit strategy configuration

Add a variable called commit_strategy and set it to the string 'auto' to represent automatic offset commits.

Kafka

consumer_config = {
    'bootstrap.servers': 'localhost:9092',
    'group.id': 'test-group',
    'auto.offset.reset': 'earliest'
}
# Set commit_strategy to 'auto' or 'manual'
# Your code here

Hint

Just create a variable named commit_strategy and assign it the string 'auto'.

3

Implement consumer offset commit logic

Write a for loop that iterates over messages. Inside the loop, if commit_strategy is 'manual', call consumer.commit() after processing each message. Assume consumer and messages are predefined.

Kafka

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

# Process messages and commit offsets if commit_strategy is 'manual'
# Your code here

Hint

Use a for loop over messages and an if statement to check commit_strategy.

4

Print committed offsets

Write a print statement to display the string 'Committed offsets:' followed by the result of consumer.committed().

Kafka

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

# Assume consumer and messages are predefined
for message in messages:
    # Process the message (placeholder)
    if commit_strategy == 'manual':
        consumer.commit()

# Print committed offsets
# Your code here

Hint

Use print('Committed offsets:', consumer.committed()) to show the committed offsets.

Practice

1. What is the main purpose of committing consumer offsets in Kafka?

easy

A. To delete old messages from the topic

B. To increase the speed of message production

C. To create new topics automatically

D. To track which messages have been processed to avoid duplicates

Solution

1. Step 1: Understand consumer offset role

   Consumer offsets mark the position of the last processed message in a partition.

2. Step 2: Identify the purpose of committing offsets

   Committing offsets helps Kafka know where to resume reading after a restart, preventing duplicate processing.

3. Final Answer:

   To track which messages have been processed to avoid duplicates -> Option D

4. Quick Check:

   Offset commit = track processed messages [OK]

Hint: Offsets mark progress to avoid reprocessing messages [OK]

Common Mistakes:

• Confusing offset commit with message production speed
• Thinking offsets delete messages
• Believing offsets create topics

2. Which Kafka consumer configuration enables automatic offset commits?

easy

A. enable.auto.commit=false

B. auto.offset.reset=earliest

C. enable.auto.commit=true

D. auto.commit.interval.ms=0

Solution

1. Step 1: Identify the setting for automatic commits

   The setting enable.auto.commit controls if Kafka commits offsets automatically.

2. Step 2: Check the correct value to enable auto commit

   Setting it to true enables automatic offset commits.

3. Final Answer:

   enable.auto.commit=true -> Option C

4. Quick Check:

   enable.auto.commit=true means auto commit enabled [OK]

Hint: Look for enable.auto.commit=true to enable auto commits [OK]

Common Mistakes:

• Confusing auto.offset.reset with auto commit
• Setting enable.auto.commit to false expecting auto commit
• Misunderstanding auto.commit.interval.ms controls commit frequency, not enabling

3. Given this code snippet, what will be the output if the consumer commits offsets manually after processing 3 messages?

consumer.subscribe(["topic"])
for i in range(5):
    msg = consumer.poll(1000)
    print(f"Processing message {i}")
    if i == 2:
        consumer.commitSync()

medium

A. Processing message 0
Processing message 1
Processing message 2
Processing message 3
Processing message 4

B. Processing message 0
Processing message 1
Processing message 2

C. Processing message 0
Processing message 1
Processing message 2
Processing message 3

D. No output due to commit error

Solution

1. Step 1: Analyze the loop and commit timing

   The loop runs 5 times, printing "Processing message i" each time from 0 to 4.

2. Step 2: Understand commitSync effect

   Calling commitSync() at i=2 commits offsets but does not stop processing.

3. Final Answer:

   Processing message 0 Processing message 1 Processing message 2 Processing message 3 Processing message 4 -> Option A

4. Quick Check:

   commitSync commits but loop continues [OK]

Hint: commitSync doesn't stop processing; loop runs fully [OK]

Common Mistakes:

• Assuming commitSync stops the loop
• Thinking only messages before commit are processed
• Confusing commitSync with commitAsync behavior

4. Identify the error in this Kafka consumer code snippet related to offset commits:

consumer.subscribe(["topic"])
while True:
    msg = consumer.poll(1000)
    if msg is not None:
        process(msg)
    consumer.commitAsync()

medium

A. poll() should not have a timeout

B. commitAsync() is called outside the message check, causing commits even when no message is received

C. process(msg) should be called after commitAsync()

D. subscribe() must be called inside the loop

Solution

1. Step 1: Review commitAsync placement

   commitAsync() is called every loop iteration, even if no message is received.

2. Step 2: Understand why this is a problem

   Committing offsets without processing messages can commit the same offset repeatedly or cause unexpected behavior.

3. Final Answer:

   commitAsync() is called outside the message check, causing commits even when no message is received -> Option B

4. Quick Check:

   commitAsync outside message check causes wrong commits [OK]

Hint: Call commit only after processing a message [OK]

Common Mistakes:

• Calling commitAsync regardless of message presence
• Moving subscribe inside the loop unnecessarily
• Misordering process and commit calls

5. You want to ensure your Kafka consumer starts reading from the earliest offset if no committed offset exists, but commits offsets manually only after processing batches of 10 messages. Which configuration and strategy combination is correct?

hard

A. Set auto.offset.reset to earliest and disable enable.auto.commit, then call commitSync() after every 10 messages

B. Set auto.offset.reset to latest and enable enable.auto.commit, commit automatically every 10 messages

C. Set auto.offset.reset to earliest and enable enable.auto.commit, commit manually after every message

D. Set auto.offset.reset to none and disable enable.auto.commit, never commit offsets

Solution

1. Step 1: Choose offset reset policy

   Setting auto.offset.reset to earliest ensures reading from the start if no offset exists.

2. Step 2: Select commit strategy

   Disabling enable.auto.commit allows manual control; calling commitSync() after 10 messages commits offsets in batches.

3. Final Answer:

   Set auto.offset.reset to earliest and disable enable.auto.commit, then call commitSync() after every 10 messages -> Option A

4. Quick Check:

   earliest + manual commit after batch = correct strategy [OK]

Hint: earliest + manual commit after batch = safe and controlled [OK]

Common Mistakes:

• Using latest offset reset when earliest is needed
• Enabling auto commit but expecting manual control
• Not committing offsets causing reprocessing