THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
| Table of Contents |
|---|
Status
Current state: Under Discussion
...
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.
...
* Some messages with earlier timestamp might also be consumed."
Proposed Changes
This process will be separated in 3 steps:
Generate Reset Plan file with specific offsets by partition to reset, for a specific Consumer Group
Verify that the Reset Plan file is consistent and the current state of Consumer Group offsets
Execute the Reset Plan file, resetting offsets to the values specified in it.
Also, we will consider an option to generate and execute in one step, for users that don't need to validate the Reset Plan file.
This options will enable clients to move back and forward, depending on where they are currently in the topic.
For instance: if a topic 'T1' with 1 partition and 100 records (offsets 0-99) has 2 Consumer Groups: one Consumer Group 'C1' inactive, with current offset 20, and other Consumer Group 'C2' that has been actively consuming records from T1. If we reset to a point in time where offset is 80, clients will be moving forward. But if C2 is reset to the same point in time, it will be moving backward.
Public Interfaces
Consumer Group Reset Offset Tool
This tool will offer 3 main operations:
--generate: This operation will generate a Reset Consumer Group Offset Reset Plan file.
--verify: This operation will validate that Consumer Group Offsets values defined in the Reset Plan file are valid in the cluster.
--execute: This operation will reset Consumer Group Offsets using values defined in the Reset Plan file.
Additionally, this tool with support an additional operation to generate and execute in the one step:
--generate-and-execute
Generate Reset Plan file
...
We are considering to add the options in ConsumerGroupCommand to reset offsets to support the following scenarios:
- Reset by time:
- 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: When we want to go back some period ago (e.g. P7D to reprocess all records from one week ago)
- Reset to Earliest: When we want to reprocess all the records available by partition.
- 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.
- Reset to Current Time: When we want to only print out and/or backup the current offset by partition.
- Reset by position:
- Reset to Offset: When we want to move to an specific offset.
- Reset to Current Offset Minus 'n' offsets: When we want to reprocess an specific number of records by partition (e.g. `minus 5` will reprocess the 5 prior records 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.)
- 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.
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
| ID | Argument | Type | Descriptio |
|---|---|---|---|
| 1. | --group | Required | Consumer Group ID. |
Scenarios
| ID | Scenario | Arguments | Description |
|---|---|---|---|
| 1.a. | Reset to Timestamp | --reset-to-datetime YYYY-MM-DDTHH:mm:SS.sss | This option will translate the datetime to Epoch milliseconds, find the offsets by timestamp, and reset to those offsets. |
| 1.b. | Reset to Period | --reset- |
...
| to-period PnYnMnDTnHnMnS |
...
Default
This process will require 3 arguments:
...
--group: defines the Consumer Group ID (i.e. ´group.id´). Required.
...
--topics: comma-separated list of topics. Optional, if it’s not specified it will use all topics consumed by Consumer Group.
| This option will subtract the period to the current timestamp in the server, and find the offsets using that subtracted timestamp, and reset to those offsets. | |||
| 1.c. | Reset to Earliest | --reset-to-earliest | This option will reset offsets to the earliest using Kafka Consumer's `auto.offset.reset` to `earliest` |
| 1.d. | Reset to Latest | --reset-to-latest | This option will reset offsets to the latest using Kafka Consumer's `auto.offset.reset` to `latest` |
| 1.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. |
| 2.a. | Reset to Offset | --reset-to | This option will reset offsets to an specific value. |
| 2.b. | Reset to Current Offset Minus 'n' offsets | --reset-minus n | This option will subtract 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 |
| 3.a | Reset from File | --reset-from-file PATH_TO_FILE | This option will take a Reset Plan file with the offsets to reset by partitions. |
Scopes
| ID | Scope | Arguments | Description |
|---|---|---|---|
| 1. | All topics | (no scope arguments) | In this case the tool will run the scenario considering all topics that has been consumed by Consumer Group. It will consider all partitions |
| 2. | List of Topics | --topics topic1,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 |
| 3. | 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 |
| 4. | Topic and List of Partitions | --topic topic1 --partitions 0,1,2 | 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. |
Execution Options
| ID | Option | Arguments | Description |
|---|---|---|---|
| 1. | Plan | (no execution arguments) | This execution option will only print out the result of the scenario by scope. |
| 2. | Execute | --execute | This execution option will run the reset offset process based on scenario and scope. |
| 3. | Export | --export path_to_file | This execution option will export the plan to a file, that later could be used to reset the offsets. (i.e. as backup) |
...
Reset Plan File: JSON Format
| Code Block | ||
|---|---|---|
| ||
{
“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
} |
Reset to Timestamp
This option will support a DateTime format (https://www.w3.org/TR/xmlschema-2/#dateTime) as input. This value will be transformed to epoch milliseconds and then used to get offsets by partition.
Pros:
This options will be the most precise using timestamp, as you move to a specific point in time.
Cons:
This option will require users to know exactly where they want to move.
Command:
´´´
bin/kafka-consumer-group-reset.sh --generate --group ConsumerGroup1 --topics topic1 --reset-to-timestamp 2017-01-01T09:00:00.000 --output kafka-consumer-group-reset.json
´´´
Reset From Period
This options will support Duration format (https://www.w3.org/TR/xmlschema-2/#duration) as input. This value will be subtracted from current timestamp from client’s machine to obtain the epoch milliseconds to be used to get offsets by partition.
Pros:
This option will be the easiest to use, as it does not require to know a specific timestamp, but give a point in time depending on when you execute the command.
Useful for debugging/development purposes where you just need to reprocess a bunch of records from an previous period.
Cons:
This option will be less accurate than previous one.
Command:
´´´
bin/kafka-consumer-group-reset.sh --generate --group ConsumerGroup1 --topics topic1 --reset-from-period P1M --output kafka-consumer-group-reset.json
´´´
Default
This option will generate a file with the current Consumer Groups offsets by partition. The result of this execution will generate a JSON file that could be used as a template to modify the offsets manually for all partitions.
This option will be useful for clusters prior to release 0.10.0.0, to create an initial file where they can specify specific offsets later. And also could be use as a fixed point in time before starting the application and later reset to this point.
Pros:
Easiest way to generate a JSON file with the right format.
Cons:
Could be cumbersome to users to modify specific offsets by partition.
Command:
´´´
bin/kafka-consumer-group-reset.sh --generate --group ConsumerGroup1 --topics topic1 --output kafka-consumer-group-reset.json
´´´
Verify JSON file
Once JSON file is generated, the tool will offer an option to validate it with the current state of the cluster. The tool will validate the following:
Topics exist
Partitions exist
Offsets exist
Latest/Current/Oldest offset
If topics or partitions don’t exist in the cluster, tool will advise that those topics/partitions won’t be processed.
If offset is not available anymore (e.g. retention remove it from the log), tool will advise that will reset the offset to the oldest or latest (i.e. resetting the offset using ‘auto.offset.reset’ policy).
Latest/Current/Oldest offset information will give the user the idea about where in the log the offset will be set.
This process will require 1 argument:
--reset-json-file: relative or absolute path to JSON file generated/customized.
Command:
´´´
bin/kafka-consumer-group-reset.sh --verify --reset-json-file kafka-consumer-group-reset.json
´´´
Execute JSON file
Once user feels comfortable with the Reset Plan file, could prepare the environment to execute the file.
To execute the reset process, the tool will require and validate that the Consumer Group become inactive first, to avoid inconsistencies in the client side.
...
--reset-json-file: relative or absolute path to JSON file generated/customized.
Command:
´´´
bin/kafka-consumer-group-reset.sh --execute --reset-json-file kafka-consumer-group-reset.json
´´´
Finally user can use the ´verify´ option to validate that current and defined offset are the same.
Compatibility, Deprecation, and Migration Plan
This KIP won’t require a Migration Plan.
...
‘reset-to-datetime’ and ‘reset-fromto-period’ will require Timestamp Index, which will make them only available from version 0.10.1.0.
...
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
...
Generate:
...
- that only one scenario is specified
...
Validate that current offsets by partition are obtained when default option is executed
...
Validate that an offset is obtained when a datetime is specified
...
Validate that oldest offset is obtained when a datetime is older than the oldest timestamp by partition
...
Validate that latest offset is obtained when a datetime is higher than the latest timestamp by partition
...
Validate that an offset is obtained when a period is specified
...
Validate that oldest offset is obtained when a period minus current timestamp is older than the oldest timestamp by partition
...
Validate that latest offset is obtained when a period minus current timestamp is higher than the latest timestamp by partition
- 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)
...
Verify:
Validate that JSON structure is correct
Validate that topics exist
Validate that partitions exist
Validate that topics are actually consumed by Consumer Group
Warn that topic does not exist and nothing will be executed
Warn that topic is not been consumed by Consumer Group and nothing will be executed
Warn that partition does not exist and nothing will be executed.
Warn that specified offset is out of range (ie. from oldest to latest) and it will be reset to oldest or latest depending on which one is closer.
...
Execute:
Validate that ConsumerGroup is inactive.
Validate that offsets are updated in cluster after execution.
Warn that non-existing/not-consumed topic has not been updated
Warn that non-existing partitions has not been updated
- Warn that offsets out of bound has been updated to oldest or latest, depending which one is closer.
Rejected Alternatives
None