How to Reset Offset in Apache Kafka?

Apache Kafka provides a robust and scalable messaging system that allows consumers to read messages from topics at specific offsets. Sometimes, you may need to reset a consumer group's offset—either to replay messages, skip messages, or troubleshoot consumption issues. For instance, if an application that processes real-time transactions fails and needs to recover lost data, resetting the offset ensures all events are replayed. Similarly, if a consumer starts lagging and processes outdated data, resetting the offset to the latest ensures only fresh messages are consumed. In this blog, we will explore various methods on how to reset offset, their use cases, and how to implement them effectively.

What is an Offset in Kafka?

An offset in Kafka is a unique identifier for messages within a partition. Each message in a partition is assigned an offset, which consumers use to track their reading position. Kafka does not automatically delete old messages based on offset but rather follows retention policies.

Why Reset an Offset?

There are several reasons to reset a consumer group’s offset:

• Replay messages: If a consumer misses processing messages due to failure, you may want to reset to an earlier offset to reprocess lost data.
• Skip bad data: If there’s corrupt or unnecessary data, resetting the offset forward helps in avoiding those messages.
• Debugging and testing: Developers might want to replay events for debugging purposes, ensuring that consumer logic behaves as expected.
• Reprocess historical data: Some applications, such as analytics or compliance audits, need to replay past messages for further analysis.
• New consumer joins a group: When a new consumer is added to a group, it might need to start consuming messages from the beginning or from the latest available offset to align with the group's behavior.
• Recover from an erroneous consumer state: If a consumer is stuck or consuming incorrectly due to an offset misconfiguration, resetting it to an appropriate value can help recover normal operations.

How to Reset Offset

Example 1: Resetting Offset to Earliest for a Consumer Group

Consider a scenario where an analytics system needs to reprocess all past data from a Kafka topic. You can reset the offset to the earliest:

kafka-consumer-groups.sh --bootstrap-server kafka-broker:9092 --group analytics-consumer --reset-offsets --to-earliest --execute

After running this command, all consumers in the analytics-consumer group will start processing messages from the beginning.

Example 2: Resetting Offset to a Specific Timestamp

Imagine a payment processing system where you want to reprocess transactions from a specific point in time (e.g., UNIX timestamp 1697500000):

kafka-consumer-groups.sh --bootstrap-server kafka-broker:9092 --group payment-group --reset-offsets --to-timestamp 1697500000 --execute

This allows you to replay only the necessary data without processing outdated transactions.

Example 3: Resetting Offset for a Specific Partition

If a specific partition of a topic is causing issues due to incorrect data, you can reset only that partition's offset:

kafka-consumer-groups.sh --bootstrap-server kafka-broker:9092 --group user-events-group --reset-offsets --to-offset 100 --topic user-events --partition 2 --execute

This is useful when you need targeted debugging rather than resetting all partitions.

Available Options to Reset an Offset

Kafka provides several ways to reset consumer offsets:

1. Using Kafka CLI (kafka-consumer-groups.sh)

Kafka’s built-in CLI tool provides an easy way to reset offsets for a consumer group.

List Consumer Groups

kafka-consumer-groups.sh --bootstrap-server <BROKER> --list

Check Current Offsets

kafka-consumer-groups.sh --bootstrap-server <BROKER> --group <GROUP_NAME> --describe

Reset Offset to the Earliest (Replay Messages)

kafka-consumer-groups.sh --bootstrap-server <BROKER> --group <GROUP_NAME> --reset-offsets --to-earliest --execute

Reset Offset to the Latest (Skip Old Messages)

kafka-consumer-groups.sh --bootstrap-server <BROKER> --group <GROUP_NAME> --reset-offsets --to-latest --execute

2. Using Kafka Consumer API

You can programmatically reset offsets using Kafka’s Consumer API.

Example: Seek to Earliest Offset

package main

import (
    "fmt"
    "time"
    "github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
    config := &kafka.ConfigMap{
        "bootstrap.servers": "localhost:9092",
        "group.id": "my-group",
        "auto.offset.reset": "earliest",
    }

    consumer, err := kafka.NewConsumer(config)
    if err != nil {
        panic(err)
    }
    defer consumer.Close()

    consumer.Subscribe("my-topic", nil)

    for {
        msg, err := consumer.ReadMessage(time.Second * 10)
        if err == nil {
            fmt.Printf("Received message: %s\n", string(msg.Value))
        }
    }
}

3. Using a Kafka Management Tool

Several Kafka management tools such as Confluent Control Center, and LinkedIn's Burrow provide graphical interfaces to reset consumer offsets. These tools make it easy to visualize consumer group status and perform offset resets with a few clicks.