KAFKA-4060 and KAFKA-4476 follow up by mjsax · Pull Request #2404 · apache/kafka

mjsax · 2017-01-19T00:24:49Z

ZK removed reveal a bug in StreamPartitionAssigner but did not fix it properly. This is a follow up bug fix.

Issue:

If topic metadata is missing, StreamPartitionAssigner should not create any affected tasks that consume topics with missing metadata.
Depending downstream tasks should not be create either.
For tasks that are not created, no store changelog topics (if any) should get created
For tasks that write output to not-yet existing internal repartitioning topics, those repartitioning topics should not get created

mjsax · 2017-01-19T00:26:33Z

Bug fix discovered by ZK removal (test was not correctly added in KAFKA-4476).

Some additional minor cleanup.

@hjafarpour @guozhangwang @enothereska @dguy

asfbot · 2017-01-19T00:30:27Z

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

asfbot · 2017-01-19T01:13:31Z

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

asfbot · 2017-01-19T03:25:30Z

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

guozhangwang · 2017-01-19T03:49:24Z

        metadataWithInternalTopics = metadata;
-        if (internalTopicManager != null)
-            metadataWithInternalTopics = metadata.withPartitions(allRepartitionTopicPartitions);
+        metadataWithInternalTopics = metadata.withPartitions(allRepartitionTopicPartitions);


Is line 372 still needed?

Good point. Will remove it.

guozhangwang · 2017-01-19T03:51:40Z

                    throw new TopologyBuilderException(String.format("stream-thread [%s] Topics not co-partitioned: [%s]", streamThread.getName(), Utils.mkString(Arrays.asList(topics), ",")));
                }
+            } else {
+                if (allRepartitionTopicsNumPartitions.get(topic).numPartitions == NOT_AVAILABLE) {


It this possible to happen? Could you elaborate this a bit?

Yes. If we have an unknown topic (c.f. the test "unknownTopic") that needs repartitioning, we don't know the number of partitions for the (to be created) internal repartitioning topic. And than this case hits.

Do we then just ignore it if it is NOT_AVAILABLE?

Yes. If metadata is missing, we skip over this and hope to get the metadata on the next refresh. But this strategy is not working properly yet. See #2404 (comment)

We might want to add a check for user-topics if they exist though -- currently, the app would just idle if a user topic does not exits.

mjsax · 2017-01-19T07:20:31Z

        final Map<String, PartitionAssignor.Assignment> assignment = partitionAssignor.assign(metadata, subscriptions);

+        final Map<String, Integer> expectedCreatedInternalTopics = new HashMap<>();
+        expectedCreatedInternalTopics.put(applicationId + "-KSTREAM-JOINOTHER-0000000012-store-changelog", 3);


Thinking about this once more, I think, there is still an issue. As we do not know the number the input topic partitions for one join input, we cannot know the number of task for the join, and thus, can actually not create this two changelog topics either.

\cc @guozhangwang WDYT?

If that was the case wouldn't we get an exception due to the topics not being co-partitioned? We should have a test for this case.

When some expected external topics do not exist or do not have metadata updatable, current strategy is that Streams will not create any tasks of its "affected" sub-topologies, including the changelog topic and repartition topic as well, and will expect another rebalance to trigger once these topics shows up. If the source topics are input ones for the "head sub-topologies" then no tasks will be created and no internal topics will be created.

As I expected. But this strategy does not work correctly...The tasks for join-subtopology should not get created (an neither this two changelog topics). Will investigate an update this PR.

@dguy After metadata refresh, we might end up with a problem. With the latest changes, we should throw an exception as number of partitions would not match the expected number. I am not sure if we can have an extra test for this -- from my understanding, this test should cover the case (and maybe other). I am not sure, if we cover all cases...

dguy · 2017-01-19T08:59:36Z

-    public final static int UNKNOWN = -1;
-    public final static int NOT_AVAILABLE = -2;
+    final static int UNKNOWN = -1;
+    final static int NOT_AVAILABLE = -2;


seeing as you are restricting visibility (thanks!) this one could be private

dguy · 2017-01-19T09:01:58Z

-
-                internalTopicManager.makeReady(topic, numPartitions);
+        for (Map.Entry<String, InternalTopicMetadata> entry : topicPartitions.entrySet()) {
+            InternalTopicConfig topic = entry.getValue().config;


maybe you could make these locals final while you are at it?

dguy · 2017-01-19T09:06:53Z

                    throw new TopologyBuilderException(String.format("stream-thread [%s] Topics not co-partitioned: [%s]", streamThread.getName(), Utils.mkString(Arrays.asList(topics), ",")));
                }
+            } else {
+                if (allRepartitionTopicsNumPartitions.get(topic).numPartitions == NOT_AVAILABLE) {


Do we then just ignore it if it is NOT_AVAILABLE?

dguy · 2017-01-19T09:11:51Z

        final Map<String, PartitionAssignor.Assignment> assignment = partitionAssignor.assign(metadata, subscriptions);

+        final Map<String, Integer> expectedCreatedInternalTopics = new HashMap<>();
+        expectedCreatedInternalTopics.put(applicationId + "-KSTREAM-JOINOTHER-0000000012-store-changelog", 3);


If that was the case wouldn't we get an exception due to the topics not being co-partitioned? We should have a test for this case.

mjsax · 2017-01-20T21:39:06Z

@guozhangwang @dguy @enothereska

Updated this.

asfbot · 2017-01-20T21:41:08Z

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

asfbot · 2017-01-20T21:41:23Z

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

asfbot · 2017-01-20T21:42:12Z

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

asfbot · 2017-01-20T21:54:48Z

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

asfbot · 2017-01-20T21:54:51Z

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

asfbot · 2017-01-20T21:55:58Z

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

do not create tasks with unkonw input partitions

asfbot · 2017-01-21T00:08:38Z

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

asfbot · 2017-01-21T00:18:39Z

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

asfbot · 2017-01-21T00:43:04Z

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

…Manager branch and fixed one copartitioning bug ZK removed reveal a bug in `StreamPartitionAssigner` but did not fix it properly. This is a follow up bug fix. Issue: - If topic metadata is missing, `StreamPartitionAssigner` should not create any affected tasks that consume topics with missing metadata. - Depending downstream tasks should not be create either. - For tasks that are not created, no store changelog topics (if any) should get created - For tasks that write output to not-yet existing internal repartitioning topics, those repartitioning topics should not get created Author: Matthias J. Sax <matthias@confluent.io> Reviewers: Damian Guy, Guozhang Wang Closes #2404 from mjsax/kafka-4060-zk-test-follow-up (cherry picked from commit 0b99bea) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2017-01-21T21:19:06Z

LGTM and merged to trunk, 0.10.2

mjsax · 2017-01-22T04:01:31Z


            if (partitions == null) {
                log.info("Skipping assigning topic {} to tasks since its metadata is not available yet", topic);
+                maxNumPartitions = StreamPartitionAssignor.NOT_AVAILABLE;


@guozhangwang I think, I missed to insert an break; here. Otherwise this only works if the "last" topic is not known. Or actually, we can replace this statement with return StreamPartitionAssignor.NOT_AVAILABLE;

Just opened a PR for this: #2418

…Manager branch and fixed one copartitioning bug ZK removed reveal a bug in `StreamPartitionAssigner` but did not fix it properly. This is a follow up bug fix. Issue: - If topic metadata is missing, `StreamPartitionAssigner` should not create any affected tasks that consume topics with missing metadata. - Depending downstream tasks should not be create either. - For tasks that are not created, no store changelog topics (if any) should get created - For tasks that write output to not-yet existing internal repartitioning topics, those repartitioning topics should not get created Author: Matthias J. Sax <matthias@confluent.io> Reviewers: Damian Guy, Guozhang Wang Closes apache#2404 from mjsax/kafka-4060-zk-test-follow-up

KAFKA-4060 and KAFKA-4476 follow up

3c59db1

guozhangwang reviewed Jan 19, 2017

View reviewed changes

mjsax commented Jan 19, 2017

View reviewed changes

dguy reviewed Jan 19, 2017

View reviewed changes

github comments

2b489a5

do not create tasks with unkonw input partitions

mjsax force-pushed the kafka-4060-zk-test-follow-up branch from a6b6194 to 2b489a5 Compare January 20, 2017 23:20

asfgit closed this in 0b99bea Jan 21, 2017

mjsax commented Jan 22, 2017

View reviewed changes

Conversation

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

Uh oh!

Uh oh!

mjsax commented Jan 19, 2017

Uh oh!

asfbot commented Jan 19, 2017

Uh oh!

asfbot commented Jan 19, 2017

Uh oh!

asfbot commented Jan 19, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

mjsax commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 20, 2017

Uh oh!

asfbot commented Jan 21, 2017

Uh oh!

asfbot commented Jan 21, 2017

Uh oh!

asfbot commented Jan 21, 2017

Uh oh!

guozhangwang commented Jan 21, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

4 participants

mjsax commented Jan 19, 2017 •

edited

Loading

mjsax Jan 22, 2017 •

edited

Loading