KAFKA-5003: StreamThread should catch InvalidTopicException by mjsax · Pull Request #2747 · apache/kafka

mjsax · 2017-03-28T01:05:43Z

We should catch InvalidTopicException and not just NoOffsetForPartitionException. Also, we need to step through all partitions that might be affected and reset those.

mjsax · 2017-03-28T01:06:19Z

Call for review @dguy @enothereska @bbejeck

asfbot · 2017-03-28T01:52:24Z

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

asfbot · 2017-03-28T01:55:22Z

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

asfbot · 2017-03-28T02:00:11Z

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

enothereska · 2017-03-28T08:46:18Z

@mjsax is there any test failing that we can use to check if this works as expected? Thanks.

bbejeck · 2017-03-30T01:37:15Z

nit: maybe change method name to addToResetList since resetting does not occur here?

bbejeck · 2017-03-30T01:45:53Z

Should we update the KStreamsFineGrainedAutoResetIntegrationTest to include a case where an out of range offset is requested?

bbejeck · 2017-03-30T01:46:13Z

@mjsax 2 minor comments, otherwise LGTM

mjsax · 2017-03-31T01:33:32Z

Updated: renamed method and added test for OffsetOutOfRangeException

\cc @enothereska @dguy @bbejeck

bbejeck · 2017-03-31T01:48:53Z

thanks @mjsax - LGTM

asfbot · 2017-03-31T02:17:09Z

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

asfbot · 2017-03-31T02:22:02Z

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

asfbot · 2017-03-31T04:31:00Z

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

guozhangwang · 2017-03-31T05:53:50Z

retest this please

asfbot · 2017-03-31T06:39:17Z

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

asfbot · 2017-03-31T06:41:47Z

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

asfbot · 2017-03-31T07:31:33Z

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

dguy

one minor comment otherwise LGTM

dguy · 2017-03-31T08:34:22Z

if (loggedTopics.add(topic)) {...}

mjsax · 2017-03-31T20:48:42Z

Updated.

asfbot · 2017-03-31T21:21:58Z

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

asfbot · 2017-03-31T21:35:06Z

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

asfbot · 2017-04-01T00:09:35Z

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

mjsax · 2017-04-02T00:26:43Z

Retest this please.

asfbot · 2017-04-02T00:33:20Z

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

asfbot · 2017-04-02T03:27:30Z

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

asfbot · 2017-04-02T03:28:31Z

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

dguy · 2017-04-03T09:40:52Z

I'm thinking this block should be extracted to a method. It doesn't really have much to do with pollRequests and it would be nicer to read if there was a method like resetPositions or similar

enothereska · 2017-04-03T10:39:31Z

@mjsax not sure why the checks are failing.

asfbot · 2017-04-03T16:30:50Z

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

asfbot · 2017-04-03T16:31:18Z

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

asfbot · 2017-04-03T16:31:20Z

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

mjsax · 2017-04-03T16:38:26Z

@eno @dguy Updated. Also updated #2774
The build is quite unstable lately -- I did rebase this branch to get latest test fixed for trunk -- the other PR did have 3 green build (before the latest update). Hope it will pass now.

asfbot · 2017-04-03T17:06:54Z

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

asfbot · 2017-04-03T17:06:58Z

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

asfbot · 2017-04-03T17:07:09Z

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

dguy

LGTM

enothereska · 2017-04-05T07:15:53Z

LGTM

ijuma

Thanks for the PR, left a couple of questions.

ijuma · 2017-04-05T08:42:19Z

-                    consumer.seekToEnd(ex.partitions());
+                    addToResetList(partition, seekToEnd, "stream-thread [%s] no custom setting defined for topic %s using original config %s for offset reset", "latest", loggedTopics);
                }
-                log.info(String.format("stream-thread [%s] no custom setting defined for topic %s using original config %s for offset reset", this.getName(), partition.topic(), originalReset));


Why did we remove this log line? And what is the behaviour if there's no reset policy defined?

It is logged in addToResetList. In StreamsConfig we default the reset policy to earliest

Having a second look, it seems that we throw an exception if the reset policy is not one of the expected values:

if (originalReset == null || (!originalReset.equals("earliest") && !originalReset.equals("latest"))) {

So the code can never reach this line.

We can reach this line: there are two places the config can be set: globally (called originalReset here) and locally per topic. Thus, this exception here, only happens if both are invalid. We check the originalReset only, if no topic-local setting was specified.

I am not sure you understood my comment. I am suggesting that the line where the log.info was removed can never be reached. Are you saying that it can be reached?

Yes. The old line (555) was hit, when no topic-level offset-strategy was specified, but a global one. Note, in Streams, we "remember" the global config but don't provide it to the consumer -- we set consumer strategy to none always to allow for different topic-level reset strategy in the first place (ie, we apply the global one "manually" within Streams)

@mjsax I think @ijuma is right in that line 555 was never reached.
I guess we're removing it anyway now.

Thanks all, I think I now understand the misunderstanding.

ijuma · 2017-04-05T08:50:57Z

-        final KStream<String, String> namedTopicsStream = builder.stream(TOPIC_Y, TOPIC_Z);
+        final KStream<String, String> pattern1Stream = builder.stream(KStreamBuilder.AutoOffsetReset.EARLIEST, Pattern.compile("topic-\\d_1"));
+        final KStream<String, String> pattern2Stream = builder.stream(KStreamBuilder.AutoOffsetReset.LATEST, Pattern.compile("topic-[A-D]_1"));
+        final KStream<String, String> namedTopicsStream = builder.stream(TOPIC_Y_1, TOPIC_Z_1);


Are these implicitly set to EARLIEST? If so, probably worth adding a comment. Same for the other test. Also, do we test the case where the original config is latest or not set?

They are set to whatever is set in StreamsConfig, default is earliest.

Don't think a comment is required. Streams people know that default is earliest.

The point is making it clear what is being tested. Anyway, you didn't reply to the second question: "Also, do we test the case where the original config is latest or not set?"

We don't have a test -- can add one.

mjsax · 2017-04-05T18:11:08Z

@ijuma Thanks for talking a look. Did reply.

mjsax · 2017-04-06T00:50:47Z

@ijuma Added one more test.

asfbot · 2017-04-06T01:37:27Z

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

asfbot · 2017-04-06T01:37:54Z

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

asfbot · 2017-04-06T01:39:01Z

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

ijuma · 2017-04-06T10:47:59Z

-        namedTopicsStream.to(stringSerde, stringSerde, DEFAULT_OUTPUT_TOPIC);
+        pattern1Stream.print();
+        pattern2Stream.print();
+        namedTopicsStream.print();


Is this intentional? cc @enothereska

Looks like it's for debugging. @mjsax ?

OK, will remove before merging.

Thx. Missed to remove it...

ijuma · 2017-04-06T11:03:59Z

LGTM, merged to trunk.

bbejeck reviewed Mar 30, 2017

View reviewed changes

mjsax force-pushed the minor-fix-reset branch from 5d0c52c to 37faac0 Compare March 31, 2017 01:30

mjsax mentioned this pull request Mar 31, 2017

KAFKA-5003: StreamThread should catch InvalidTopicException #2774

Closed

dguy reviewed Mar 31, 2017

View reviewed changes

dguy reviewed Apr 3, 2017

View reviewed changes

mjsax changed the title ~~MINOR: StreamThread should catch InvalidTopicException~~ KAFKA-5003: StreamThread should catch InvalidTopicException Apr 3, 2017

mjsax added 3 commits April 3, 2017 09:28

MINOR: StreamThread should catch InvalidTopicException

d1a4a0a

Github comments and rebasing

e846ada

Damian's comment

a4bdf4d

mjsax force-pushed the minor-fix-reset branch from c13436d to 7af4ded Compare April 3, 2017 16:28

Damian's comment

ac3c9ee

mjsax force-pushed the minor-fix-reset branch from 7af4ded to ac3c9ee Compare April 3, 2017 16:35

dguy approved these changes Apr 4, 2017

View reviewed changes

ijuma reviewed Apr 5, 2017

View reviewed changes

added one more test

fee0a4f

ijuma reviewed Apr 6, 2017

View reviewed changes

asfgit closed this in afeadbe Apr 6, 2017

mjsax deleted the minor-fix-reset branch April 7, 2017 03:18

Conversation

mjsax commented Mar 28, 2017

Uh oh!

mjsax commented Mar 28, 2017

Uh oh!

asfbot commented Mar 28, 2017

Uh oh!

asfbot commented Mar 28, 2017

Uh oh!

asfbot commented Mar 28, 2017

Uh oh!

enothereska commented Mar 28, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck commented Mar 30, 2017

Uh oh!

bbejeck commented Mar 30, 2017

Uh oh!

mjsax commented Mar 31, 2017

Uh oh!

bbejeck commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

guozhangwang commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Mar 31, 2017

Uh oh!

asfbot commented Apr 1, 2017

Uh oh!

mjsax commented Apr 2, 2017

Uh oh!

asfbot commented Apr 2, 2017

Uh oh!

asfbot commented Apr 2, 2017

Uh oh!

asfbot commented Apr 2, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

enothereska commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

mjsax commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

asfbot commented Apr 3, 2017

Uh oh!

dguy left a comment

Choose a reason for hiding this comment

Uh oh!

enothereska commented Apr 5, 2017