Table of Contents

Status

Current state: Under DiscussionAccepted

Discussion thread: here

JIRA: KAFKA-4743

Jira

server	ASF JIRA
columns	key,summary,type,created,updated,due,assignee,reporter,priority,status,resolution
serverId	5aa69414-a9e9-3523-82ec-879b028fb15b
key	KAFKA-4743

Released: 0.11.0.0

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

...

Proposed Changes

We are considering to add the options in ConsumerGroupCommand to reset offsets to support the following scenarios:

Reset by time:

.

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:

Reset to Datetime:

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)

Reset to Period:

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.

Reset by position:

6.	Reset to Offset

:

When we want to move to an specific offset.

Reset to Current Offset Minus

7.	Shift Offset by 'n'

offsets:

When we want to

reprocess an specific number of records by partition (e.g. `minus 5` will reprocess the 5 prior records

move forward or backward from the current

Consumer Group offset by partition).

Reset to Current Offset Plus 'n' offsets: When we want to avoid process an specific number of records after the current Consumer Group offset by partition (e.g. `plus 5` will move the Current Group offset 5 position 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

Reset by file

From exported plan:

This operations will be executed by Consumer Group and will have 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.

And there will be 3 execution options:

Plan: Print the result of the operation (i.e. show probable offsets to reset if the operation is executed) (Default)
Execute: Users will have to explicitly ask to execute the reset tool.
Export: Export plan to a JSON file to execute it later.

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

Public Interfaces

Consumer Group Reset Offset options

Main option

--reset-offsetoffsets

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

...

and it will only support new consumers.

Required Arguments

ID	Argument	Type	Descriptio
1.	--group	Required	Consumer Group ID.
2.	--bootstrap-server	Required	Server to connect to.

Scenarios

At least one of the scenarios should be defined to proceed with the execution

Reset to Period

ID	Scenario	Arguments	ConsiderationsDescription	Example
1.a.	Reset to Timestampto Datetime	--to-datetime YYYY-MM-DDTHH:mm:SS.sss±hh:mm --to-datetime YYYY-MM-DDTHH:mm:SS.sssZ - --reset -to-datetime YYYY-MM-DDTHH:mm:SS.sss	Datetime must be specified in ISO8601 format. This option will translate the datetime to Epoch milliseconds, find the offsets by timestamp, and reset to those offsets.	1.b.	If the Timezone is not specified, it will use UTC.	Reset to first offset since 01 January 2017, 00:00:00 hrs	--reset-offsets –group test.group --topic foo --to-datetime 2017-01-01T00:00:00Z
2.	Reset by Duration	--by-duration -period PnYnMnDTnHnMnS	Duration must be specified in ISO8601 format. This option will subtract the period duration to the current timestamp in the server, and find the offsets using that subtracted timestamp, and reset to those offsets. The duration specified won't consider daylight saving effects.	Reset to first offset since one week ago (from current timestamp): --reset-offsets --group test.group --topic foo --by-duration P7D
31.c.	Reset to Earliest	--reset-to-earliest	This option will reset offsets to the earliest using Kafka Consumer's `auto.offset.reset` to `earliest``#seekToBeginning`	Reset to earliest offset available: --reset-offsets --group test.group --topic foo --to-earliest
4.1.d.	Reset to Latest	--reset-to-latest	This option will reset offsets to the latest earliest using Kafka Consumer's `auto.offset.reset` to `latest``#seekToEnd`	Reset to latest offset available: --reset-offsets --group test.group --topic foo --to-latest
51.e.	Reset to Current Position	(no scenario arguments)	This option won't reset the offset. It will be used to print and export current offset.	Reset to current position: --reset-offsets --group test.group --topic foo
62.a.	Reset to Offset	--resetto-tooffset	This option will reset offsets to an specific value.	Reset to offset 1 in all partitions: --reset-offsets --group test.group --topic foo --to-offset 1
7.	Shift Offset by 'n'	--shift-by	2.b.	Reset to Current Offset Minus 'n' offsets	--reset-minus n	This option will subtract add the `n` value to the current offset, and reset to the result.
2.c.	Reset to Current Offset Plus 'n' offsets	--reset-plus n	This option will add the `n` value to the current offset, and reset to the result
`n` can be a positive or negative value, so offset will be move backward if it is negative, and forward if it is positive. If current offset + n is higher than the latest offset, new offset will be set to latest. If current offset + n is lower than the earliest offset, new offset will be set to earliest.	Reset to current offset plus 5 positions: --reset-offsets --group test.group –topic foo --shift-by 5
8	3.a	Reset from File	--reset-from-file PATH_TO_FILE	This option will take a Reset Plan CSV file with the offsets to reset by topics/partitions. It does not require scope, because topics and partitions are defined in the file.	Reset using a file with reset plan: --reset-offsets --group test.group --from-file reset-plan.csv

Scopes

The scopes will be defined extending the existing `topic` argument, using the following format:

Code Block
--topic <topic-name>:<partition numbers> ex: --topic topic1 --topic topic2:0,1,2

Topic all partitions

ID	Scope	Arguments	Description	Example
1.	All topics(no scope arguments)	--all-topics	In this case the tool will run the scenario considering all topics that has been consumed by Consumer Group. It will consider all partitions by topic assigned.	Reset offsets in all topics and partitions: --reset-offsets --group test.group --all-topics –to-earliest
2.	List of Topics	--topicstopic topic1 ,topic2...--topic topic2	In this case, the tool first will validate that input topics are been consumed by Consumer Group, and then run the scenario. It will consider all partitions by topic defined.	Reset offsets on all partitions of a list of topics: --reset-offsets --group test.group --topic foo --topic bar –to-earliest
3.	Topics and List of Partitions	--topic topic1:0,1,2 --topic topic2:0	In this case, the tool first will validate that input topic is been consumed by Consumer Group, and then run the scenario. It will consider	only the partitions specified by topic.	Reset offsets on specific partitions by topics: --reset-offsets --group test.group --topic foo:1,2 --topic bar:	4.	Topic and List of Partitions	--topic topic1 --partitions 0,1,2,3 –to-earliest
4.	Topic	--topic topic1	In this case, the tool first will validate that input topic is been consumed by Consumer Group, and then run the scenario. It will consider all partitions.	Reset offsets on a specifc topic, all partitions; --reset-offsets --group test.group --topic foo --to-earliest only the partitions specified.

Execution Options

ID

Option

Arguments

Description

Examples

1.

Plan

(no execution arguments)

This execution option will only print out the result of the scenario by scope.

The output will look like this:

TOPIC                 PARTITION NEW-OFFSET NEW-LAG LOG-END-OFFSET CONSUMER-ID HOST CLIENT-ID
foo                   0         90         10      100            -           -    -

Prints result of resetting all topics and partitions to earliest:

--reset-offsets --group test.group --all-topics –to-earliest

2.

Execute

--execute

This execution option will run the reset offset process based on scenario and scope.

Prints and execute resetting all topics and partitions to earliest:

--reset-offsets --group test.group --all-topics –to-earliest --execute

3.

Export

--export path_to_file

This execution option will export print out the plan to a filereset plan in CSV format, that later could be used to reset in the offsetsscenario 8. (i.e. as backup)

Prints plan for resetting all topics and partitions to earliest in CSV format:

--reset-offsets --group test.group --all-topics –to-earliest --export

Execute and Prints plan for resetting all topics and partitions to earliest in CSV format:

--reset-offsets --group test.group --all-topics –to-earliest --export --execute

Reset Plan File:

...

CSV Format

The CSV schema will consist on:

<topic>,<partition>,<offset>

Sample:

topic-1,0,100
topic-1,1,200
topic-1,2,0
topic-2,0,0

Implementation Details

These options will be implemented inside the `ConsumerGroupCommand`. The `resetOffsets` operation will looks like this:

Code Block

language	js	scala

def resetOffsets(): Map[PartitionAssignmentState, Long] = {
   “offsets”: {
    “group_id”: “ConsumerGroup1”,
    “topics”: [
	  “topic1”: [val groupId = opts.options.valueOf(opts.groupOpt)
  val (state, assignments) = describeGroup() //(1)
  assignments match {
    case None =>
      // applies to both old and new consumer
      printError(s"The consumer group '$groupId' does not exist.")
      Map.empty
    case Some(assignments) =>
      state match {
        case  “partition”: 0,
		  “offset”: 0
	    },
		{Some("Dead") =>
          printError(s"Consumer group '$groupId' does not exist.")
          Map.empty
        case Some("Empty") => //(2)
          val assignmentsToReset = getAssignmentsToReset(assignments) //(3)
          “partition”: 1,
		  “offset”: 0
	    }
 val assignmentsPrepared = prepareAssignmentsToReset(assignmentsToReset) //(4)
          val execute = opts.options.has(opts.executeOpt)
          if(execute) //(5)
            ],
	  “topic2”: [resetAssignments(assignmentsPrepared)
          assignmentsPrepared
        {case Some("PreparingRebalance") | Some("AwaitingSync") =>
          “partition”: 0,
		  “offset”: 0
	    },
		{
 printError(s"Consumer group '$groupId' offsets cannot be reset if it is rebalancing.")
          Map.empty
        case Some("Stable") =>
          printError(s"Consumer group '$groupId'  “partition”: 1,
		  “offset”: 0
	    }
offsets cannot be reset if it has members active.")
          Map.empty
        case other =>
          ]// the control should never reach here
    ]
  },
  “version”: 1
}      throw new KafkaException(s"Expected a valid consumer group state, but found '${other.getOrElse("NONE")}'.")
      }
  }
}

(1) It will get assignments from `describeGroup` operation

(2) Then use `getAssignmentsToReset` to filter the assignments with the values defined by `–topic` of or `--all-topics`.

(3) Then get new offset by assignments using `prepareAssignmentsToReset`.

(4) It will be only executed when the ConsumerGroup selected is inactive o avoid race conditions.

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

To change the offsets.

Compatibility, Deprecation, and Migration Plan

...

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.

This KIP will not support old consumers that store offsets in Zookeeper.

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)

...

Space shortcuts

Child pages

Versions Compared

Old Version 7

New Version Current

Key

Status

Proposed Changes

Public Interfaces

Consumer Group Reset Offset options

Main option

Required Arguments

Scenarios

Scopes

Execution Options

Reset Plan File:

CSV Format

Implementation Details

Compatibility, Deprecation, and Migration Plan

Test Plan

Space shortcuts

Child pages

Page History

Versions Compared

Old Version 7

New Version Current

Key

Status

Proposed Changes

Public Interfaces

Consumer Group Reset Offset options

Main option

Required Arguments

Scenarios

Scopes

Execution Options

Reset Plan File:

CSV Format

Implementation Details

Compatibility, Deprecation, and Migration Plan

Test Plan