KAFKA-4588: Wait for topics to be created in QueryableStateIntegrationTest.shouldNotMakeStoreAvailableUntilAllStoresAvailable by dguy · Pull Request #2371 · apache/kafka

dguy · 2017-01-13T18:11:14Z

After debugging this i can see the times that it fails there is a race between when the topic is actually created/ready on the broker and when the assignment happens. When it fails StreamPartitionAssignor.assign(..) gets called with a Cluster with no topics. Hence the test hangs as no tasks get assigned. To fix this I added a waitForTopics method to EmbeddedKafkaCluster. This will wait until the topics have been created.

asfbot · 2017-01-13T18:12:00Z

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

dguy · 2017-01-13T18:12:15Z

@guozhangwang, @enothereska, @mjsax

asfbot · 2017-01-13T18:59:01Z

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

asfbot · 2017-01-13T18:59:29Z

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

guozhangwang · 2017-01-13T19:38:08Z

I have a clarification question about the issue, which is not directly related to this test failure: in practice, it is likely that StreamsPartitionAssignor.assign() gets called when the passed in Cluster does not include all the refreshed metadata, if it has been created just before the starting of the app. If it does cause the assignor to fail I think it is a general issue to fix.

So when you say test hangs as no tasks gets assigned, which line of code is specifically causing this issue, is it because of this:

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java

Line 305 in 4b71c0b

do {

dguy · 2017-01-13T20:51:56Z

@guozhangwang it isn't an issue with StreamsPartitionAssignor.assign() - it doesn't fail. It just receives an empty Cluster object and then does the right thing, i.e., assigns nothing. So the StreamThread.ConsumerRebalanceListener.onPartitionsAssigned gets called with an empty assignment and no tasks get created.

What happen is we call KafkaConsumer.subscribe(..) it in turn calls metadata.setTopics(..). This triggres a metadata refresh. The Cluster metadata that comes back and is set in metadata.update(..) is empty. This is due to:
WARN Error while fetching metadata with correlation id 2 : {stream-three-2=UNKNOWN_TOPIC_OR_PARTITION} (org.apache.kafka.clients.NetworkClient:695)
Then when ConsumerCoordinator.performAssign happens and it calls metadata.fetch() it gets the empty Cluster object and passes it to StreamPartitionAssignor.assign(...)

The assignment never happens again (at least not within the 30 seconds the test runs for)

mjsax · 2017-01-13T21:01:03Z

This might fix queryOnRebalance(), too. Did start to investigate but could not find a cause so far...

asfbot · 2017-01-13T21:12:32Z

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

guozhangwang

LGTM overall, left a single comment.

guozhangwang · 2017-01-13T22:51:15Z

    }
+
+    public void waitForTopics(final StreamsConfig config, final String...topics) throws InterruptedException, IOException {
+        final StreamsKafkaClient streamsKafkaClient = new StreamsKafkaClient(config);


Instead of using a separate client, could we migrate the scala TestUtils.waitUntilMetadataIsPropagated code, that looks into broker.apis.metadataCache directly?

by migrate I mean we can add a similar function in the java TestUtils class.

sure i'll take a look at it.

Hmm, can't add it the the java TestUtils class as clients doesn't have a dependency on core.
Thought i'd add it to StreamsTestUtils, but can't as it breaks checkstyle:

[ant:checkstyle] /Users/damian/dev/apache/kafka/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:18:1: Disallowed import - kafka.api.PartitionStateInfo. [ant:checkstyle] /Users/damian/dev/apache/kafka/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:19:1: Disallowed import - kafka.api.Request. [ant:checkstyle] /Users/damian/dev/apache/kafka/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:20:1: Disallowed import - kafka.server.KafkaServer. [ant:checkstyle] /Users/damian/dev/apache/kafka/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:21:1: Disallowed import - kafka.server.MetadataCache. [ant:checkstyle] /Users/damian/dev/apache/kafka/streams/src/test/java/org/apache/kafka/test/StreamsTestUtils.java:26:1: Disallowed import - scala.Option.

Ok - i've moved it to IntegrationTestUtils as the org.apache.kafka.streams.integration package is already allowed to import most of these. I've had to add 1 more package to the allowed imports.

…eated

asfbot · 2017-01-16T10:24:09Z

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

asfbot · 2017-01-16T11:11:39Z

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

asfbot · 2017-01-16T13:18:46Z

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

guozhangwang · 2017-01-16T18:49:26Z

retest this please

guozhangwang · 2017-01-16T18:50:02Z

LGTM overall. Seems ResetIntegrationTest test failure has some other causes (seen this in Jenkins builds, cc @mjsax ).

asfbot · 2017-01-16T18:56:25Z

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

asfbot · 2017-01-16T19:37:53Z

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

guozhangwang · 2017-01-16T21:13:31Z

Is this test failure relevant? I thought it should be gone as well with the fix: https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/918/testReport/junit/org.apache.kafka.streams.integration/JoinIntegrationTest/testInnerKStreamKTable/

asfbot · 2017-01-16T21:50:06Z

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

dguy · 2017-01-16T22:08:15Z

No. I haven't changed that test

…

On Mon, 16 Jan 2017 at 21:50, asfbot ***@***.***> wrote: Refer to this link for build results (access rights to CI server needed): https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/920/ Test FAILed (JDK 8 and Scala 2.11). — You are receiving this because you authored the thread. Reply to this email directly, view it on GitHub <#2371 (comment)>, or mute the thread <https://github.com/notifications/unsubscribe-auth/AADWLSJYW8J3By8Hb0yFnc_jLoUC8LIdks5rS-YYgaJpZM4LjJls> .

guozhangwang · 2017-01-17T04:05:06Z

@dguy Do you think it makes sense to always enforce waitForTopicPartitions in createTopics, or there are some scenarios that we do not want to wait for that?

dguy · 2017-01-17T09:36:38Z

@guozhangwang Sure makes sense. I'll do that

asfbot · 2017-01-17T12:01:29Z

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

asfbot · 2017-01-17T12:22:36Z

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

asfbot · 2017-01-17T12:27:50Z

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

dguy · 2017-01-17T13:42:49Z

retest this please

asfbot · 2017-01-17T14:31:57Z

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

asfbot · 2017-01-17T14: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/931/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-17T16:01:11Z

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

…nTest.shouldNotMakeStoreAvailableUntilAllStoresAvailable After debugging this i can see the times that it fails there is a race between when the topic is actually created/ready on the broker and when the assignment happens. When it fails `StreamPartitionAssignor.assign(..)` gets called with a `Cluster` with no topics. Hence the test hangs as no tasks get assigned. To fix this I added a `waitForTopics` method to `EmbeddedKafkaCluster`. This will wait until the topics have been created. Author: Damian Guy <damian.guy@gmail.com> Reviewers: Matthias J. Sax, Guozhang Wang Closes #2371 from dguy/integration-test-fix (cherry picked from commit 825f225) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2017-01-17T20:33:46Z

Thanks @dguy . Merged to trunk and piggy-backed to 0.10.2.

mjsax · 2017-01-17T20:56:45Z

@guozhangwang ResetIntegrationTest fails from time to time since ZK dependency got removed. Maybe it is related to this. But as ZK PR still has some follow up tasks, I am not sure if we should talk any action for the test already of just monitor it the next days.

…nTest.shouldNotMakeStoreAvailableUntilAllStoresAvailable After debugging this i can see the times that it fails there is a race between when the topic is actually created/ready on the broker and when the assignment happens. When it fails `StreamPartitionAssignor.assign(..)` gets called with a `Cluster` with no topics. Hence the test hangs as no tasks get assigned. To fix this I added a `waitForTopics` method to `EmbeddedKafkaCluster`. This will wait until the topics have been created. Author: Damian Guy <damian.guy@gmail.com> Reviewers: Matthias J. Sax, Guozhang Wang Closes apache#2371 from dguy/integration-test-fix

dguy added 3 commits January 13, 2017 16:51

metadata hack

52f5792

wait for topics to be created

c85face

revert StreamPartitionAssignor

d392046

mjsax approved these changes Jan 13, 2017

View reviewed changes

guozhangwang reviewed Jan 13, 2017

View reviewed changes

guozhangwang mentioned this pull request Jan 15, 2017

KAFKA-4633: Always using regex pattern subscription in StreamThread #2379

Closed

look in broker metadata cache to see if topic/partitions have been cr…

ef7a75f

…eated

wait for topic to be created in createTopic

83ed16c

asfgit closed this in 825f225 Jan 17, 2017

dguy deleted the integration-test-fix branch January 19, 2017 11:40

Conversation

dguy commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

dguy commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

guozhangwang commented Jan 13, 2017

Uh oh!

dguy commented Jan 13, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax commented Jan 13, 2017

Uh oh!

asfbot commented Jan 13, 2017

Uh oh!

guozhangwang left a comment

Choose a reason for hiding this comment

Uh oh!

guozhangwang Jan 13, 2017

Choose a reason for hiding this comment

Uh oh!

guozhangwang Jan 13, 2017

Choose a reason for hiding this comment

Uh oh!

dguy Jan 14, 2017

Choose a reason for hiding this comment

Uh oh!

dguy Jan 16, 2017

Choose a reason for hiding this comment

Uh oh!

dguy Jan 16, 2017

Choose a reason for hiding this comment

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

guozhangwang commented Jan 16, 2017

Uh oh!

guozhangwang commented Jan 16, 2017

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

guozhangwang commented Jan 16, 2017

Uh oh!

asfbot commented Jan 16, 2017

Uh oh!

dguy commented Jan 16, 2017 via email

Uh oh!

guozhangwang commented Jan 17, 2017

Uh oh!

dguy commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

dguy commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

asfbot commented Jan 17, 2017

Uh oh!

guozhangwang commented Jan 17, 2017 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Uh oh!

mjsax commented Jan 17, 2017

dguy commented Jan 13, 2017 •

edited

Loading

guozhangwang commented Jan 17, 2017 •

edited

Loading