KAFKA-4733: Improve Streams Reset Tool console output #2503

gwenshap · 2017-02-05T04:28:09Z

Added general explanation of the tool and what it does. Also added few details to the arguments.

asfbot · 2017-02-05T04:30:29Z

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

asfbot · 2017-02-05T04:30:46Z

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

asfbot · 2017-02-05T04:31:00Z

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

asfbot · 2017-02-05T17:59:38Z

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

asfbot · 2017-02-05T18:08:19Z

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

mjsax

Thanks for the PR Gwen!

mjsax · 2017-02-05T18:38:12Z

core/src/main/scala/kafka/tools/StreamsResetter.java

            .withRequiredArg()
            .ofType(String.class)
            .withValuesSeparatedBy(',')
            .describedAs("list");
-        intermediateTopicsOption = optionParser.accepts("intermediate-topics", "Comma-separated list of intermediate user topics")
+        intermediateTopicsOption = optionParser.accepts("intermediate-topics", "Comma-separated list of intermediate user topics (topics created using through() method). For these topics, the tool will skip to the end.")


topics created using through() method -> topics used in/by through() method

mjsax · 2017-02-05T18:41:53Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+                "*** Important! You will get wrong output if you don't clean up the statestore after running the reset tool!\n\n"
+        );
+        parser.printHelpOn(System.err);
+


Nit: remove empty line ;)

mjsax · 2017-02-05T18:43:01Z

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -271,6 +271,18 @@ private boolean isInternalTopic(final String topicName) {
            && (topicName.endsWith("-changelog") || topicName.endsWith("-repartition"));
    }

+    private void printHelp(OptionParser parser) throws IOException {
+        System.err.println("The Application Reset Tool allows you to quickly reset an application in order to reprocess its data from scratch.\n" +


Can you please format this differently. No line should be longer than 120 chars and please start a new line after each sentence (otherwise reviewing is quite cumbersome). Thx

mjsax · 2017-02-05T18:43:38Z

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -271,6 +271,18 @@ private boolean isInternalTopic(final String topicName) {
            && (topicName.endsWith("-changelog") || topicName.endsWith("-repartition"));
    }

+    private void printHelp(OptionParser parser) throws IOException {
+        System.err.println("The Application Reset Tool allows you to quickly reset an application in order to reprocess its data from scratch.\n" +
+                "* This tool will reset offsets of input topics to 0 and it skips to the end of intermediate topics (topics created with the through() method).\n" +


topics created with the through() method
as above

I would remove will: This tool reset[s] offsets
same below

mjsax · 2017-02-05T18:44:21Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+        System.err.println("The Application Reset Tool allows you to quickly reset an application in order to reprocess its data from scratch.\n" +
+                "* This tool will reset offsets of input topics to 0 and it skips to the end of intermediate topics (topics created with the through() method).\n" +
+                "* This tool will deletes the internal topics that were created automatically (topics ending in -changelog and -repartition).\n  You do not need to specify internal topics the tool finds them automatically.\n" +
+                "* This tool will not delete output topics (if you want to delete them, you need to do it yourself through bin/kafka-topics.sh command)\n" +


. at the end missing

mjsax · 2017-02-05T18:44:46Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+                "* This tool will reset offsets of input topics to 0 and it skips to the end of intermediate topics (topics created with the through() method).\n" +
+                "* This tool will deletes the internal topics that were created automatically (topics ending in -changelog and -repartition).\n  You do not need to specify internal topics the tool finds them automatically.\n" +
+                "* This tool will not delete output topics (if you want to delete them, you need to do it yourself through bin/kafka-topics.sh command)\n" +
+                "* This tool will does not clean up the local state on the stream application instances (the persisted stores used to cache aggregation results).\n  You need to call KafkaStreams#cleanUp() in your application or manually delete them from the directory specified by state.dir configuration (usually /tmp/kafka-streams/<application.id>\n\n" +


typo: will does not

usually -> default
(I hope that people usually change the default to avoid unwanted state recreating if temp gets wiped out :)

And: closing ) missing at the end.

"state.dir"

mjsax · 2017-02-05T18:47:24Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+                "* This tool will deletes the internal topics that were created automatically (topics ending in -changelog and -repartition).\n  You do not need to specify internal topics the tool finds them automatically.\n" +
+                "* This tool will not delete output topics (if you want to delete them, you need to do it yourself through bin/kafka-topics.sh command)\n" +
+                "* This tool will does not clean up the local state on the stream application instances (the persisted stores used to cache aggregation results).\n  You need to call KafkaStreams#cleanUp() in your application or manually delete them from the directory specified by state.dir configuration (usually /tmp/kafka-streams/<application.id>\n\n" +
+                "*** Important! You will get wrong output if you don't clean up the statestore after running the reset tool!\n\n"


statestore -> local stores (to use same terminology as above)

mjsax · 2017-02-05T18:50:00Z

core/src/main/scala/kafka/tools/StreamsResetter.java

+    private void printHelp(OptionParser parser) throws IOException {
+        System.err.println("The Application Reset Tool allows you to quickly reset an application in order to reprocess its data from scratch.\n" +
+                "* This tool will reset offsets of input topics to 0 and it skips to the end of intermediate topics (topics created with the through() method).\n" +
+                "* This tool will deletes the internal topics that were created automatically (topics ending in -changelog and -repartition).\n  You do not need to specify internal topics the tool finds them automatically.\n" +


automatically by Kafka Streams (maybe remove "automatically"?)

maybe: topics starting with "<application.id>-" ?

[because] the tool

asfbot · 2017-02-05T19:14:57Z

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

asfbot · 2017-02-06T00:17:40Z

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

asfbot · 2017-02-06T00:24:19Z

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

asfbot · 2017-02-06T00:48:53Z

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

mjsax

LGTM

\cc any of @miguno @enothereska @dguy for second review

miguno · 2017-02-06T18:57:08Z

core/src/main/scala/kafka/tools/StreamsResetter.java

@@ -138,12 +138,12 @@ private void parseArguments(final String[] args) throws IOException {
            .ofType(String.class)
            .defaultsTo("localhost:2181")
            .describedAs("url");
-        inputTopicsOption = optionParser.accepts("input-topics", "Comma-separated list of user input topics")
+        inputTopicsOption = optionParser.accepts("input-topics", "Comma-separated list of user input topics. For these topics, the tool will reset the offset to 0.")


Nit: I think this is not always true -- it will reset to the earliest available offset, which may not necessarily be 0, right? (same comment applies to the other mention of offset 0 in the CLI help output below).

I'll fix this, but note that I copied this explanation out of Confluent's docs, so we may need to fix those too :)

Good point. @mjsax could you take care of updating the Confluent docs accordingly?

asfbot · 2017-02-07T22:30:32Z

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

asfbot · 2017-02-07T22:31:22Z

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

asfbot · 2017-02-07T22:33:39Z

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

mjsax · 2017-02-08T01:54:56Z

@guozhangwang Call for final review and merging.

gwenshap · 2017-02-08T06:10:59Z

I feel like I'll need to start paying the review debt for @guozhangwang soon...

guozhangwang

LGTM and merged to trunk.

Added general help and some explanations of arguments

04465a1

style fixes

01be320

mjsax requested changes Feb 5, 2017

View reviewed changes

format changes and improvements suggested by Matthias

f3e23b5

mjsax approved these changes Feb 6, 2017

View reviewed changes

miguno reviewed Feb 6, 2017

View reviewed changes

improved accuracy of explanation per request from Michael

8073dca

asfgit closed this in 2e662a0 Feb 8, 2017

guozhangwang reviewed Feb 8, 2017

View reviewed changes

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4733: Improve Streams Reset Tool console output #2503

KAFKA-4733: Improve Streams Reset Tool console output #2503

gwenshap commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

mjsax left a comment

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

mjsax Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 6, 2017

asfbot commented Feb 6, 2017

asfbot commented Feb 6, 2017

mjsax left a comment

miguno Feb 6, 2017

gwenshap Feb 7, 2017

miguno Feb 7, 2017

asfbot commented Feb 7, 2017

asfbot commented Feb 7, 2017

asfbot commented Feb 7, 2017

mjsax commented Feb 8, 2017

gwenshap commented Feb 8, 2017

guozhangwang left a comment

KAFKA-4733: Improve Streams Reset Tool console output #2503

KAFKA-4733: Improve Streams Reset Tool console output #2503

Conversation

gwenshap commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

asfbot commented Feb 5, 2017

mjsax 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

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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Feb 5, 2017

asfbot commented Feb 6, 2017

asfbot commented Feb 6, 2017

asfbot commented Feb 6, 2017

mjsax 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

asfbot commented Feb 7, 2017

asfbot commented Feb 7, 2017

asfbot commented Feb 7, 2017

mjsax commented Feb 8, 2017

gwenshap commented Feb 8, 2017

guozhangwang left a comment

Choose a reason for hiding this comment