Follow-up improvements for consumer offset reset tool (KIP-122) #3102

jeqo · 2017-05-19T14:50:57Z

Implement improvements defined here: https://issues.apache.org/jira/browse/KAFKA-5266

asfbot · 2017-05-19T16:12:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4188/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-19T16:19:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4175/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-05-19T21:59:51Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4195/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-05-19T22:00:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4208/
Test PASSed (JDK 7 and Scala 2.11).

hachikuji

Thanks for the patch. Left a few comments.

hachikuji · 2017-05-22T22:17:06Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

      } else if (opts.options.has(opts.topicOpt)) {
        val topics = opts.options.valuesOf(opts.topicOpt).asScala
-        parseTopicPartitionsToReset(topics)
+        val topicPartitions = parseTopicPartitionsToReset(topics)
+        allTopicPartitions.filter(topicPartition => topicPartitions.contains(topicPartition))


I'm not sure I get why we filter based on current committed offsets. Why would we not permit the tool to set offsets for a topic partition which hasn't been previously consumed? If that is really the behavior we want, I would prefer to raise an exception rather than silently proceeding with only some of the offsets.

I haven't considered setting offsets on topic partitions unassigned previously, sounds like a good idea (e.g. set initial offsets for a new consumer group). if we add this, checking previous assignments won't be required, right? And implementation will be cleaner. Maybe we can add this as part of this PR?

Yes, I think it is a reasonable feature to implement here and users might find it a little surprising if they couldn't do it.

hachikuji · 2017-05-22T22:19:04Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

-            if (resetPlan.keySet.contains(topicPartition))
-              (topicPartition, resetPlan(topicPartition))
-            else null
+        resetPlan.keySet.map { topicPartition =>


Adding a comment here for lack of a better place. It seems we still require either --all-topics or --topic to use the CSV option?

hachikuji · 2017-05-22T22:20:08Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -650,6 +646,23 @@ object ConsumerGroupCommand extends Logging {
      }
    }

+    private def checkOffsetRange(topicPartition: TopicPartition, shiftedOffset: Long) = {


nit: shiftedOffset is no longer correct since we use this in more scenarios.

Agree, I'll change it to offset

hachikuji · 2017-05-22T22:20:55Z

core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

@@ -650,6 +646,23 @@ object ConsumerGroupCommand extends Logging {
      }
    }

+    private def checkOffsetRange(topicPartition: TopicPartition, shiftedOffset: Long) = {
+      val newOffset = getLogEndOffset(topicPartition) match {


nit: we don't really need the newOffest variable. We can return from the match directly.

Also, we don't handle errors in getLogEndOffset, which seems a bit unsafe. Maybe we should check for the Unknown and Ignore cases and raise an error. In general, this is the point I was trying to make: rather than crossing our fingers and trying to go ahead with the reset, It is safer to fail fast on errors.

You're right. This should also apply to the nulls returnedon #prepareOffsetsToReset, and instead return #printUsageAndDie. Do you agree?

asfbot · 2017-05-25T17:02:37Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4396/
Test FAILed (JDK 7 and Scala 2.11).

asfbot · 2017-05-25T17:49:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4382/
Test FAILed (JDK 8 and Scala 2.12).

…/KAFKA-5266 # Conflicts: # core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

asfbot · 2017-05-25T20:17:21Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4392/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-05-25T20:18:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4406/
Test PASSed (JDK 7 and Scala 2.11).

…docs updated.

asfbot · 2017-05-26T16:21:57Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4431/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-05-26T16:30:59Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4445/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-29T15:46:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4525/
Test PASSed (JDK 7 and Scala 2.11).

asfbot · 2017-05-29T15:57:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4510/
Test PASSed (JDK 8 and Scala 2.12).

ijuma · 2017-05-31T06:43:56Z

@hachikuji @jeqo What is the status of this PR?

hachikuji · 2017-05-31T07:11:26Z

@ijuma Was planning to look at it shortly.

hachikuji

Thanks for the updates. LGTM. There is some commented out code that I will remove when merging.

…P-122) Implement improvements defined here: https://issues.apache.org/jira/browse/KAFKA-5266 Author: Jorge Quilcate Otoya <quilcate.jorge@gmail.com> Reviewers: Jason Gustafson <jason@confluent.io> Closes #3102 from jeqo/feature/KAFKA-5266 (cherry picked from commit ef95512) Signed-off-by: Jason Gustafson <jason@confluent.io>

DXist · 2017-12-12T15:29:04Z

Hi. One more improvement to the tool - return nonzero code when operation is failed. For example, when there are consumers in the group and reset cannot be done.

hachikuji · 2017-12-12T16:59:25Z

@DXist Thanks for the suggestion. Would you mind opening a JIRA? https://issues.apache.org/jira/projects/KAFKA

DXist · 2017-12-13T09:12:08Z

@hachikuji Here you go: https://issues.apache.org/jira/browse/KAFKA-6357

implement improvements (1,2,3)

f021973

jeqo changed the title ~~Implement Improvements~~ Follow-up improvements for consumer offset reset tool (KIP-122) May 19, 2017

fix NPE on unassigned topic partitions

f1a4c52

add reset-offsets 'to-current' scenario

d30aaa8

hachikuji reviewed May 22, 2017

View reviewed changes

remove returned value

06c4972

jeqo added 2 commits May 25, 2017 21:18

remove returned value

ad3a18c

Merge remote-tracking branch 'origin/feature/KAFKA-5266' into feature…

93365fc

…/KAFKA-5266 # Conflicts: # core/src/main/scala/kafka/admin/ConsumerGroupCommand.scala

support not only assigned topic partition but any specified by user. …

574d284

…docs updated.

Merge branch 'trunk' into feature/KAFKA-5266

4b8093b

hachikuji approved these changes May 31, 2017

View reviewed changes

asfgit closed this in ef95512 May 31, 2017

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Follow-up improvements for consumer offset reset tool (KIP-122) #3102

Follow-up improvements for consumer offset reset tool (KIP-122) #3102

jeqo commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

hachikuji left a comment

hachikuji May 22, 2017

jeqo May 24, 2017

hachikuji May 26, 2017

hachikuji May 22, 2017

hachikuji May 22, 2017

jeqo May 24, 2017 •

edited

hachikuji May 22, 2017

hachikuji May 22, 2017

jeqo May 24, 2017

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 26, 2017

asfbot commented May 26, 2017

asfbot commented May 29, 2017

asfbot commented May 29, 2017

ijuma commented May 31, 2017

hachikuji commented May 31, 2017

hachikuji left a comment

DXist commented Dec 12, 2017

hachikuji commented Dec 12, 2017

DXist commented Dec 13, 2017

Follow-up improvements for consumer offset reset tool (KIP-122) #3102

Follow-up improvements for consumer offset reset tool (KIP-122) #3102

Conversation

jeqo commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

asfbot commented May 19, 2017

hachikuji left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

jeqo May 24, 2017 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 25, 2017

asfbot commented May 26, 2017

asfbot commented May 26, 2017

asfbot commented May 29, 2017

asfbot commented May 29, 2017

ijuma commented May 31, 2017

hachikuji commented May 31, 2017

hachikuji left a comment

Choose a reason for hiding this comment

DXist commented Dec 12, 2017

hachikuji commented Dec 12, 2017

DXist commented Dec 13, 2017

jeqo May 24, 2017 •

edited