Status

Current state: Under DiscussionAccepted

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

Released: 0.11.0.0

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 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.

...

Public Interfaces

Consumer Group Reset Offset options

...

This option should be executed independently from other Consumer Group options (list, describe, delete, etc.) and it will only support new consumers.

Required Arguments

Scenarios

Scenarios

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

...

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

Reset Plan File: CSV Format

The CSV schema will consist on:

<topic>,<partition>,<offset>

Sample:

...

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:

def resetOffsets(): Map[PartitionAssignmentState, Long] = {
  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 Some("Dead") =>
          printError(s"Consumer group '$groupId' does not exist.")
          Map.empty
        case Some("Empty") => //(2)
          val assignmentsToReset = getAssignmentsToReset(assignments) //(3)
          val assignmentsPrepared = prepareAssignmentsToReset(assignmentsToReset) //(4)
          val execute = opts.options.has(opts.executeOpt)
          if(execute) //(5)
            resetAssignments(assignmentsPrepared)
          assignmentsPrepared
        case Some("PreparingRebalance") | Some("AwaitingSync") =>
          printError(s"Consumer group '$groupId' offsets cannot be reset if it is rebalancing.")
          Map.empty
        case Some("Stable") =>
          printError(s"Consumer group '$groupId' offsets cannot be reset if it has members active.")
          Map.empty
        case other =>
          // the control should never reach here
          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:

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)

...