Proposed Changes

We are considering to add the options in ConsumerGroupCommand to reset offsets.

There will be 3 execution options available:

Plan: Print out the result of the operation (i.e. show a list of topic, partition, current offset, new offset to reset if the operation is executed). This will be the default option.
Execute: Users will have to explicitly ask to execute the reset operation depending on the scenario and the scope specified.
Export: Export plan to a CSV file to execute it later.

All these options are considered to be executed affecting only one Consumer Group.

The following scenarios will be supported:

ID	Scenario	Description
1.	Reset to Datetime	When we want to reset offsets to an specific point in time. (e.g. to 1/1/2017 at 00:00 to reprocess all records from these year)
2.	Reset from Duration	When we want to go back some period ago (e.g. P7D to reprocess all records from one week ago)
3.	Reset to Earliest	When we want to reprocess all the records available by partition.
4.	Reset to Latest	When a Consumer Groups has an offset lag and we don't want to process missing records, only move to the latest.
5.	Reset to Current Time	When we want to only print out and/or backup the current offset by partition.
6.	Reset to Offset	When we want to move to an specific offset.
7.	Shift Offset by 'n'	When we want to move forward or backward from the current position. 'n' can be a positive or negative value that will be add it to the current offset to move to a new position.
8.	Reset from file	When we have a file with the required offsets by topic/partition to reset to.

And the following scopes:

All Topics consumed by Consumer Group: This scope will consider all the topics that has been consumed by a Consumer Group.
Specific List of Topics: This scope will consider all the topics defined by user.
One Topic, All Partitions: This scope will consider only one topic and all partitions.
One Topic, Specific Partition: This scope will consider only one topic and partition specified by user.

You can check the concept-proof implementation of this feature on this branchpull request.

Public Interfaces

...

(5) It will be executed only if it is asked explictly. This will consist in create a Consumer using the same `group.id` and use:

Code Block

language	scala

consumer.assign(List(topicPartition).asJava)
consumer.seek(topicPartition, offset)
consumer.commitSync()

...