KAFKA-6318: StreamsResetter should return non-zero return code on error #4305

shivsantham · 2017-12-08T18:08:34Z

If users specify a non-existing topic as input parameter, StreamsResetter only prints out an error message that the topic was not found, but return code is still zero. We should return a non-zero return code for this case.

tedyu · 2017-12-08T18:11:23Z

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

@@ -316,6 +317,7 @@ private void maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consu
                for (final String topic : notFoundInputTopics) {
                    System.out.println("Topic: " + topic);
                }
+                topicNotFound = 1;


You reuse 1 for EXIT_CODE_ERROR.
Assigning EXIT_CODE_ERROR directly would be clearer.

mjsax · 2017-12-08T19:25:16Z

Thanks for the PR! We should add a test for this to ResetIntegrationTest

shivsantham · 2017-12-11T16:43:14Z

@mjsax & @tedyu Thanks for the review. Will add a test and make the proposed changes.

mjsax

Thanks for the PR!

We should add a test for this to ResetIntegrationTest.

mjsax · 2017-12-17T04:52:45Z

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

@@ -248,7 +249,7 @@ private void maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consu

        if (inputTopics.size() == 0 && intermediateTopics.size() == 0) {
            System.out.println("No input or intermediate topics specified. Skipping seek.");
-            return;
+            return 0;


use EXIT_CODE_SUCCESS

mjsax · 2017-12-17T04:53:17Z

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

        final List<String> inputTopics = options.valuesOf(inputTopicsOption);
        final List<String> intermediateTopics = options.valuesOf(intermediateTopicsOption);
+        int topicNotFound = 0;


init with EXIT_CODE_SUCCESS

mjsax · 2017-12-17T04:53:32Z

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

            System.err.println("ERROR: Resetting offsets failed.");
            throw e;
        }
        System.out.println("Done.");
+        return topicNotFound;
+


nit: remove empty line.

mjsax · 2017-12-17T04:53:52Z

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

@@ -326,10 +328,13 @@ private void maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consu
            }

        } catch (final Exception e) {
+


nit: remove empty line

shivsantham · 2017-12-17T21:23:33Z

@mjsax I don't seem to see the test component in my fork. Looks like i don't have the latest code. I will send out a seperate PR for the test.

mjsax · 2017-12-18T18:44:22Z

Please add the test here and include in this PR (otherwise, we might drop adding a test on the floor): https://github.com/apache/kafka/blob/trunk/streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

Thanks.

mjsax

Thanks for the update. The added test is a good test, but does not test what it is supposed to test...

I would keep and rename the test to shouldNotAllowToResetWhileStreamsIsRunning and add a two new tests for non-existing topics (one test for input topic, one test for intermediate topic). I think, for this two new test, we don't even need to create/start/stop a KafakStreams application at all.

Last comment. I don't think we need to add this new test to AbstractResetIntegrationTest but should put the code directly int ResetIntegerationTest

mjsax · 2017-12-19T18:31:40Z

streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java

+        final List<String> parameterList = new ArrayList<>(
+                Arrays.asList("--application-id", APP_ID + testNo,
+                        "--bootstrap-servers", bootstrapServers,
+                        "--input-topics", INPUT_TOPIC));


You need to set a non-existing topic name here.

Sure will change it.

mjsax · 2017-12-19T18:31:57Z

streams/src/test/java/org/apache/kafka/streams/integration/AbstractResetIntegrationTest.java

+        KafkaStreams streams = new KafkaStreams(setupTopologyWithoutIntermediateUserTopic(), streamsConfiguration);
+        streams.start();
+
+        final int exitCode = new StreamsResetter().run(parameters, cleanUpConfig);


I assume this fails, because KS is still running. This is a valid test, but does test if StreamsResetter fails for non-existing topic.

Let me change the input-topics in the parameterlist. Then i believe this should fail for sure.

shivsantham · 2017-12-21T02:47:09Z

Will add more tests without kafka streams.

mjsax

LGTM.

Call for second review @bbejeck @dguy @guozhangwang

bbejeck · 2017-12-30T01:27:29Z

streams/src/test/java/org/apache/kafka/streams/integration/ResetIntegrationTest.java

@@ -81,4 +92,44 @@ public void testReprocessingFromDateTimeAfterResetWithoutIntermediateUserTopic()
    public void testReprocessingByDurationAfterResetWithoutIntermediateUserTopic() throws Exception {
        super.testReprocessingByDurationAfterResetWithoutIntermediateUserTopic();
    }
+
+    @Test
+    public void testInputTopicAbsentWithoutIntermediateUserTopic() throws Exception {


maybe a nit: I understand what you are doing here, but IMHO there should be two separate tests.

This test just needs a rename: it calls shouldNotAllowToResetWhileStreamsIsRunning() and should have the same name. There are two separate test below: one for invalid input topic and one for invalid intermediate topic. (or did you mean something else, @bbejeck )

@mjsax @bbejeck I renamed the method. Apologize for my poor method naming skills.

No need to apologize. :)

@ssanthalingam thanks for updating, no apologies necessary!

bbejeck · 2017-12-30T01:27:58Z

one minor comment, otherwise LGTM.

guozhangwang · 2018-01-02T18:46:07Z

LGTM. Merged to trunk.

KAFKA-6318: StreamsResetter should return non-zero return code on error

a9696d0

tedyu reviewed Dec 8, 2017

View reviewed changes

Fix based on suggestions

47a9dd9

mjsax reviewed Dec 17, 2017

View reviewed changes

Minor changes: Test addition/Pending

74d06b9

Adding Test

8341d66

mjsax reviewed Dec 19, 2017

View reviewed changes

More tests & fixes

e328423

mjsax approved these changes Dec 27, 2017

View reviewed changes

bbejeck reviewed Dec 30, 2017

View reviewed changes

Fix method name

4536903

guozhangwang merged commit fd2bdbf into apache:trunk Jan 2, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6318: StreamsResetter should return non-zero return code on error #4305

KAFKA-6318: StreamsResetter should return non-zero return code on error #4305

shivsantham commented Dec 8, 2017

tedyu Dec 8, 2017 •

edited

mjsax commented Dec 8, 2017

shivsantham commented Dec 11, 2017 •

edited

mjsax left a comment

mjsax Dec 17, 2017

mjsax Dec 17, 2017

mjsax Dec 17, 2017

mjsax Dec 17, 2017

shivsantham commented Dec 17, 2017 •

edited

mjsax commented Dec 18, 2017

mjsax left a comment

mjsax Dec 19, 2017

shivsantham Dec 21, 2017

mjsax Dec 19, 2017

shivsantham Dec 21, 2017

shivsantham commented Dec 21, 2017

mjsax left a comment

bbejeck Dec 30, 2017

mjsax Dec 30, 2017

ssanthalingam Dec 30, 2017

mjsax Dec 30, 2017

bbejeck Jan 1, 2018 •

edited

bbejeck commented Dec 30, 2017

guozhangwang commented Jan 2, 2018

		@@ -326,10 +328,13 @@ private void maybeResetInputAndSeekToEndIntermediateTopicOffsets(final Map consu
		}

		} catch (final Exception e) {

KAFKA-6318: StreamsResetter should return non-zero return code on error #4305

KAFKA-6318: StreamsResetter should return non-zero return code on error #4305

Conversation

shivsantham commented Dec 8, 2017

tedyu Dec 8, 2017 • edited

Choose a reason for hiding this comment

mjsax commented Dec 8, 2017

shivsantham commented Dec 11, 2017 • edited

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

shivsantham commented Dec 17, 2017 • edited

mjsax commented Dec 18, 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

shivsantham commented Dec 21, 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

bbejeck Jan 1, 2018 • edited

Choose a reason for hiding this comment

bbejeck commented Dec 30, 2017

guozhangwang commented Jan 2, 2018

tedyu Dec 8, 2017 •

edited

shivsantham commented Dec 11, 2017 •

edited

shivsantham commented Dec 17, 2017 •

edited

bbejeck Jan 1, 2018 •

edited