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

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

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

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()

Hint

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

Practice

1. What is the main purpose of offset management in Kafka?

easy

A. To track which messages have been read by a consumer

B. To store the actual message data in Kafka topics

C. To manage the number of partitions in a topic

D. To control the replication factor of Kafka brokers

Solution

1. Step 1: Understand offset management role

   Offset management keeps track of the last message a consumer has read in a partition.

2. Step 2: Differentiate from other Kafka features

   Storing messages, partition count, and replication are unrelated to offset tracking.

3. Final Answer:

   To track which messages have been read by a consumer -> Option A

4. Quick Check:

   Offset management = track read messages [OK]

Hint: Offset management tracks read messages, not data storage [OK]

Common Mistakes:

• Confusing offset with message storage
• Mixing offset with partition or replication management
• Thinking offset controls message content

2. Which of the following is the correct way to manually commit offsets in Kafka consumer API?

easy

A. consumer.offsetCommit()

B. consumer.commitOffset()

C. consumer.saveOffset()

D. consumer.commitSync()

Solution

1. Step 1: Recall Kafka consumer API methods

   The official method to commit offsets manually is commitSync() or commitAsync().

2. Step 2: Identify the exact method name

   Only commitSync() matches the correct Kafka API method for synchronous commit.

3. Final Answer:

   consumer.commitSync() -> Option D

4. Quick Check:

   Manual commit method = commitSync() [OK]

Hint: Manual commit uses commitSync() or commitAsync() [OK]

Common Mistakes:

• Using incorrect method names like commitOffset()
• Confusing commitSync() with offsetCommit()
• Assuming saveOffset() exists

3. Given the following code snippet, what will be the output if the consumer seeks to offset 5 and reads one message?

consumer.assign(Collections.singleton(topicPartition));
consumer.seek(topicPartition, 5);
ConsumerRecords records = consumer.poll(Duration.ofMillis(1000));
for (ConsumerRecord record : records) {
    System.out.println(record.offset());
    break;
}

medium

A. 5

B. 0

C. 1

D. Exception thrown

Solution

1. Step 1: Understand seek behavior

   Calling seek sets the consumer's position to offset 5 in the assigned partition.

2. Step 2: Poll and print offset

   The poll fetches messages starting at offset 5; printing record.offset() outputs 5.

3. Final Answer:

   5 -> Option A

4. Quick Check:

   seek to 5 then poll = offset 5 [OK]

Hint: seek sets start offset; first polled message offset matches [OK]

Common Mistakes:

• Assuming poll starts from offset 0
• Expecting offset 1 instead of 5
• Thinking seek causes an exception

4. What is wrong with the following code snippet for committing offsets manually?

consumer.poll(Duration.ofMillis(1000));
consumer.commitAsync();
consumer.close();

medium

A. consumer.close() should be called before commitAsync()

B. commitAsync() should be called before poll()

C. commitAsync() is called without processing polled records

D. poll() duration is too short for commitAsync()

Solution

1. Step 1: Check offset commit timing

   Offsets should be committed after processing records, not just after poll().

2. Step 2: Identify missing processing step

   The code commits offsets without handling any messages, so it commits offsets that may not be processed.

3. Final Answer:

   commitAsync() is called without processing polled records -> Option C

4. Quick Check:

   Commit after processing, not just after poll [OK]

Hint: Always process records before committing offsets [OK]

Common Mistakes:

• Calling commit before processing messages
• Closing consumer before committing
• Misunderstanding poll duration effect

5. You want to restart a Kafka consumer and continue reading from the last committed offset, but also have the option to start from the beginning if no offset is found. Which approach correctly manages offsets to achieve this?

hard

A. Always call consumer.seekToEnd() before polling messages

B. Use consumer.seekToBeginning() if no committed offset exists, else start from committed offset

C. Manually set offset to 0 every time before polling

D. Disable offset commits and rely on auto.offset.reset = latest

Solution

1. Step 1: Understand offset reset behavior

   If no committed offset exists, consumer can seek to beginning to read all messages.

2. Step 2: Combine offset check and seek

   Check for committed offset; if found, start there; if not, call seekToBeginning() to start fresh.

3. Final Answer:

   Use consumer.seekToBeginning() if no committed offset exists, else start from committed offset -> Option B

4. Quick Check:

   Seek beginning if no offset, else resume from committed [OK]

Hint: Seek beginning only if no committed offset found [OK]

Common Mistakes:

• Always seeking to end loses old messages
• Resetting offset to 0 blindly every time
• Disabling commits causes loss of progress