Offset management in Kafka - Practice Problems & Coding Challenges

Challenge - 5 Problems

❓ Predict Output

intermediate

What is the output of this Kafka consumer offset commit code?

Consider the following Kafka consumer code snippet that commits offsets synchronously. What will be printed if the commit succeeds?

Kafka

consumer.subscribe(['my_topic'])
try:
    records = consumer.poll(timeout_ms=1000)
    for record in records:
        print(f"Received: {record.value}")
    consumer.commit()
    print("Offsets committed successfully")
except Exception as e:
    print(f"Commit failed: {e}")

A No output, consumer.poll returns empty

B Commit failed: OffsetOutOfRangeException

C Received: <message_value>\nOffsets committed successfully

D Offsets committed successfully

💡 Hint

The commit() call happens after messages are received and printed.

The consumer polls messages and prints each received message's value. After processing, commit() commits the offsets synchronously. If successful, the message 'Offsets committed successfully' is printed. So the output includes the received message(s) followed by the commit success message.

🧠 Conceptual

intermediate

Which offset reset policy causes the consumer to start from the earliest offset when no committed offset exists?

Kafka consumers can be configured with an offset reset policy. Which of the following values for 'auto.offset.reset' causes the consumer to read from the earliest offset if no committed offset is found?

A "auto"

B "latest"

C "none"

D "earliest"

💡 Hint

Think about starting from the beginning of the topic.

Setting 'auto.offset.reset' to 'earliest' tells the consumer to start reading from the earliest available offset if no committed offset exists. 'latest' starts from the end, 'none' throws an error, and 'auto' is not a valid value.

❓ Predict Output

advanced

What error does this Kafka consumer code raise?

Given the following code snippet, what error will be raised when running it?

Kafka

consumer.assign([TopicPartition('my_topic', 0)])
consumer.seek(TopicPartition('my_topic', 0), 100)
records = consumer.poll(timeout_ms=1000)
for record in records:
    print(record.value)

A TimeoutException: Poll timed out

B No error, prints messages starting from offset 100

C OffsetOutOfRangeException

D IllegalStateException: Consumer not subscribed to topic

💡 Hint

Assigning partitions manually allows seeking to offsets.

Using assign() manually assigns partitions, so the consumer is not subscribed but can seek offsets. If offset 100 exists, poll returns messages from there without error. No subscription error occurs because assign() is used instead of subscribe().

❓ Predict Output

advanced

What is the output of this asynchronous offset commit example?

Examine this Kafka consumer code that commits offsets asynchronously with a callback. What will be printed if the commit fails?

Kafka

def on_commit(err, partitions):
    if err:
        print(f"Commit failed for partitions: {partitions}")
    else:
        print("Commit succeeded")

consumer.subscribe(['my_topic'])
consumer.poll(timeout_ms=1000)
consumer.commit_async(callback=on_commit)

A Commit failed for partitions: [TopicPartition(topic='my_topic', partition=0)]

B Commit succeeded

C No output, commit_async is non-blocking

D SyntaxError due to callback parameter

💡 Hint

The callback prints a message only if there is an error.

If the commit fails, the callback receives an error and prints the failure message with the partitions. If it succeeds, it prints 'Commit succeeded'. The question asks about the failure case, so the failure message is printed.

🧠 Conceptual

expert

How many committed offsets are stored if a consumer group consumes 3 partitions and commits offsets only for 2 partitions?

A Kafka consumer group reads from 3 partitions of a topic. It commits offsets only for 2 partitions and never commits for the third. How many committed offsets will Kafka store for this group?

A 2 committed offsets, only for the partitions committed

B 1 committed offset, the latest among all partitions

C 3 committed offsets, one per partition

D 0 committed offsets, because all partitions must be committed together

💡 Hint

Offsets are stored per partition independently.

Kafka stores committed offsets per partition per consumer group. If offsets are committed only for 2 partitions, only those 2 offsets are stored. The third partition has no committed offset until committed.

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