Status

Current state: Under Discussion

Discussion thread: here 

JIRA: KAFKA-4743 

Please keep the discussion on the mailing list rather than commenting on the wiki (wiki discussions get unwieldy fast).

Motivation

This KIP wants to introduce a command-line tool to enable users to reset Consumer Group offsets.

Today, if users want to reprocess old records (i.e. records where offset is smaller than current consumer offset), they will need to modify client-side code (eg. using "KafkaConsumer#seek()" operations).

This process will involve write their own program which starts a consumer and commits offset or change client-side code, redeploy the application, and eventually rollback changes in application to consume records as usual. This is a cumbersome process given that users have to be aware of how many partitions are by topic, in which offset they are, and to which offset they want to move to.

Bringing the option to reset offsets from outside of the application will enable a cleaner and consistent way to achieve reprocessing and move along the topic.

Currently, there are 2 indexes that allow users to reprocess records: offset index and timestamp index.

Offset index represent a pointer over the sequential id that each record has by partition, available from Kafka inception.

Timestamp index (added in release 0.10.1.0 by KIP-33) represents a pointer over the epoch milliseconds from record creation, that could be defined by the Producer or the Broker, depending in the Timestamp Type.

From KIP-33:

"Searching by timestamp will have better accuracy. The guarantees provided are:

* The messages whose timestamp are after the searched timestamp will be consumed.

* Some messages with earlier timestamp might also be consumed."

Public Interfaces

Consumer Group Reset Offset options

Main option

--reset-offset

This option should be executed independently from other Consumer Group options (list, describe, delete, etc.)

Additional Arguments

Scenarios

Scopes

Execution Options

Reset Plan File: JSON Format

{
  “offsets”: {
    “group_id”: “ConsumerGroup1”,
    “topics”: [
	  “topic1”: [
        {
          “partition”: 0,
		  “offset”: 0
	    },
		{
          “partition”: 1,
		  “offset”: 0
	    }
      ],
	  “topic2”: [
        {
          “partition”: 0,
		  “offset”: 0
	    },
		{
          “partition”: 1,
		  “offset”: 0
	    }
      ]
    ]
  },
  “version”: 1
}

Compatibility, Deprecation, and Migration Plan

This KIP won’t require a Migration Plan.

‘reset-to-datetime’ and ‘reset-to-period’ will require Timestamp Index, which will make them only available from version 0.10.1.0.

Nevertheless, it is possible to achieve the same result since release 0.10.0 using timestamp metadata by record, doing a linear search starting from latest. However, we are not considering to support this option because it will be an expensive/slow option.

All the other options will be available for releases from 0.10.0.0 given that KIP-97 is implemented.

 Test Plan

• A unit test to validate that --reset-offset is executed independently from other Consumer Group options (list, delete, describe, etc.)
• A unit test to validate that only one scenario is specified
• A unit test to validate that only one scope is specified
• A unit test to validate that only one execution option is specified
• A unit test by combination of scenario, scope and execution option.
• A unit test to validate that when calculated offset by partition is older that the earliest offset, tool resets offset to earliest (e.g. when we specified --reset-to 0, when earliest offset is 10)
• A unit test to validate that when calculated offset by partition is bigger that the latest offset, tool resets offset to latest (e.g. when we specified --reset-to 0, when earliest offset is 10)

Rejected Alternatives

None

(no scenario arguments)