KAFKA-6391 ensure topics are created with correct partitions BEFORE building the… #4347

cvaliente · 2017-12-20T15:52:29Z

ensure topics are created with correct partitions BEFORE building the metadata for our stream tasks

First ensureCoPartitioning() on repartitionTopicMetadata before creating allRepartitionTopicPartitions

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

… metadata for our stream tasks

cvaliente · 2017-12-21T13:45:35Z

@guozhangwang Could you have a look on this as well since you seem to have contributed a lot to this class before?

mjsax · 2017-12-26T23:14:17Z

@cvaliente Thx for the PR. Can you elaborate on this change? The return value of copartitionGroups() as well as repartitionTopicMetadata and metadata should be the same before or after the for loop. What is the issue you try to address and why does this PR fix it?

cvaliente · 2017-12-27T08:48:27Z

@mjsax ensureCopartitioning doesn't have a return value (a pattern I'm not particular fond of). It modifies the input parameter repartitionTopicMetadata - increasing the number of partitions for copartitioned topics when necessary.

allRepartitionTopicPartitions is built from the information in repartitionTopicMetadata and used to generate all the streams task.

previously, allRepartitionTopicPartitions was built before running ensureCopartitioning on repartitionTopicMetadata, so any extra partitions added by ensureCopartitioning did not have a streams task generated from them - those partitions were being written to, but no data read from.

cvaliente · 2017-12-27T08:51:32Z

With this PR, we first validate our repartitionTopicMetadata to make sure number of partitions is correct, then we create the topics & partitions if needed, and then we create the tasks for those topics & partitions.

mjsax

Thanks for the explanation. I missed the fact that we indeed modify repartitionTopicMetadata. Good catch and thanks for the patch!

LGMT.

Call for second review @bbejeck @dguy @guozhangwang

bbejeck · 2017-12-27T20:22:17Z

Thanks for the patch @cvaliente. LGTM, but I think we should have an integration test for this.

The test should fail with the old logic, because: While stream-partition-assignor-test-KSTREAM-MAP-0000000001-repartition is created correctly with four partitions, the StreamPartitionAssignor will only assign three tasks to the topic. Test passes with the new logic.

tedyu · 2017-12-28T11:08:26Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+            new TopicPartition("topic1", 0),
+            new TopicPartition("topic1", 1),
+            new TopicPartition("topic1", 2),
+            new TopicPartition("topic3", 0),


Consider naming the topic "topic2" since there are only two topics in the test

I am using the source topics defined in the test class for this. Since topic2 has the same number of partitions as topic1, it is not suitable for this test case.

cvaliente · 2017-12-28T11:13:01Z

@bbejeck @mjsax I added a test case covering this.
The old logic fails when checking the assigned tasks, the new logic passes the test.

mjsax

Thanks for adding a test!

mjsax · 2017-12-28T21:16:49Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        // joining the stream and the table
+        // this triggers the enforceCopartitioning() routine in the StreamPartitionAssignor,
+        // forcing the stream.map to get repartitioned to a topic with four partitions.
+        stream1.join(table1,


nit: move table1 to next line, or fix indention below to align with table1

mjsax · 2017-12-28T21:17:02Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        final UUID uuid = UUID.randomUUID();
+        final String client = "client1";
+
+        mockTaskManager(Collections.<TaskId>emptySet(),


nit: as above

mjsax · 2017-12-28T21:18:04Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        subscriptions.put(
+            client,
+            new PartitionAssignor.Subscription(
+                Collections.singletonList("unknownTopic"),


why do we put "unknownTopic" here? Should we subscribe to "topic1" and "topic3"? Or we can actually pass in an empty list?

I guess this is copied from shouldNotLoopInfinitelyOnMissingMetadataAndShouldNotCreateRelatedTasks -- there we need to pass in unknownTopic as this topic does not exist in cluster metadata.

mjsax · 2017-12-28T21:22:51Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        expectedCreatedInternalTopics.put(applicationId + "-topic3-STATE-STORE-0000000002-changelog", 4);
+
+        // check if all internal topics were created as expected
+        assertThat(mockInternalTopicManager.readyTopics, equalTo(expectedCreatedInternalTopics));


Nit: assertThat take expected result as first parameter IIRC (otherwise error message on failing test is "reversed")

mjsax · 2017-12-28T21:23:16Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        );
+
+        // check if we created a task for all expected topicpartitions.
+        assertThat(new HashSet<>(assignment.get(client).partitions()), equalTo(new HashSet<>(expectedAssignment)));


as above. (please fix in shouldNotLoopInfinitelyOnMissingMetadataAndShouldNotCreateRelatedTasks, too)

cvaliente · 2017-12-29T09:22:00Z

@mjsax thanks for the feedback, done!

bbejeck · 2017-12-29T18:51:29Z

@cvaliente thanks for adding the test, LGTM.

mjsax

Sorry for the forth and back :(

mjsax · 2017-12-29T23:47:10Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+                    return new KeyValue<>(key, value);
+                }
+            })
+            .count(Materialized.<Object, Long, KeyValueStore<Bytes, byte[]>>as("count"));


We actually don't need to name the store. This could be .count() plus updating the name for the repartitioning and changelog topic.

know the naming was not required. My thinking is it would be preferable to explicitly choose the name so I don't have to rely on internal logic for naming the repartition topic that I later refer to. Unfortunately there's no explicit naming for the streams.map() repartition topic so it is all moot anyway.

I understand your point -- however, naming must be deterministic and should not change from release to release (otherwise, upgrading hard harder and we need to mention it in the upgrade docs). Thus, if we use internal names in tests, we have some implicit testing that naming does not change. That's why I prefer to not name the operator, too, if not required.

Yes I guess that's ok. I agree that an upgrade should not require a reset of the streams application's topology unless absolutely necessary. However I think in that case we should also have some explicit testing for the naming of itnernal topics (not sure if those exist yet?)

Either way I will fix this PR up once I am back in the office.

mjsax · 2017-12-29T23:48:47Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        expectedCreatedInternalTopics.put(applicationId + "-topic3-STATE-STORE-0000000002-changelog", 4);
+
+        // check if all internal topics were created as expected
+        assertThat(expectedCreatedInternalTopics, equalTo(mockInternalTopicManager.readyTopics));


Sorry for the forth and back -- for assertThat you original code was correct and expected argument is second one... (it different for assertEquals -- my bad(!)).

mjsax · 2017-12-29T23:49:07Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

+        );
+
+        // check if we created a task for all expected topicpartitions.
+        assertThat(new HashSet<>(expectedAssignment), equalTo(new HashSet<>(assignment.get(client).partitions())));


as above: flip arguments

mjsax · 2017-12-29T23:49:51Z

.../src/test/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignorTest.java

-                               Collections.<TaskId>emptySet(),
-                               uuid1,
-                builder);
+        mockTaskManager(


Thanks for the cleanup! Really appreciated!

cvaliente · 2018-01-02T08:44:24Z

@mjsax @bbejeck Cleaned up all the things that were commented on.

guozhangwang · 2018-01-02T18:22:37Z

LGTM! Merged to trunk.

ensure topics are created with correct partitions BEFORE building the…

bda1803

… metadata for our stream tasks

mjsax reviewed Dec 27, 2017

View reviewed changes

Clemens Valiente added 3 commits December 28, 2017 10:11

add a few explanations in the comments

f54c9e4

remove extra newline

940c823

tedyu reviewed Dec 28, 2017

View reviewed changes

mjsax reviewed Dec 28, 2017

View reviewed changes

Clemens Valiente added 2 commits December 29, 2017 10:19

Fix intendations, argument order in asserts and c&p error

f055b97

fixed one intendation that I missed.

6834aee

mjsax reviewed Dec 29, 2017

View reviewed changes

fix assertThat parameters and remove materialization from count()

bf5cfaf

guozhangwang merged commit 5ca1226 into apache:trunk Jan 2, 2018

cvaliente deleted the KAFKA-6391 branch January 2, 2018 22:44

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-6391 ensure topics are created with correct partitions BEFORE building the… #4347

KAFKA-6391 ensure topics are created with correct partitions BEFORE building the… #4347

cvaliente commented Dec 20, 2017

cvaliente commented Dec 21, 2017

mjsax commented Dec 26, 2017

cvaliente commented Dec 27, 2017 •

edited

Loading

cvaliente commented Dec 27, 2017

mjsax left a comment

bbejeck commented Dec 27, 2017

tedyu Dec 28, 2017

cvaliente Dec 28, 2017

cvaliente commented Dec 28, 2017

mjsax left a comment

mjsax Dec 28, 2017

mjsax Dec 28, 2017

mjsax Dec 28, 2017

mjsax Dec 28, 2017

mjsax Dec 28, 2017

cvaliente commented Dec 29, 2017

bbejeck commented Dec 29, 2017

mjsax left a comment

mjsax Dec 29, 2017

cvaliente Dec 30, 2017

mjsax Dec 30, 2017

cvaliente Dec 31, 2017

mjsax Dec 29, 2017

mjsax Dec 29, 2017

mjsax Dec 29, 2017

cvaliente commented Jan 2, 2018

guozhangwang commented Jan 2, 2018

KAFKA-6391 ensure topics are created with correct partitions BEFORE building the… #4347

KAFKA-6391 ensure topics are created with correct partitions BEFORE building the… #4347

Conversation

cvaliente commented Dec 20, 2017

Committer Checklist (excluded from commit message)

cvaliente commented Dec 21, 2017

mjsax commented Dec 26, 2017

cvaliente commented Dec 27, 2017 • edited Loading

cvaliente commented Dec 27, 2017

mjsax left a comment

Choose a reason for hiding this comment

bbejeck commented Dec 27, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

cvaliente commented Dec 28, 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

cvaliente commented Dec 29, 2017

bbejeck commented Dec 29, 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

cvaliente commented Jan 2, 2018

guozhangwang commented Jan 2, 2018

cvaliente commented Dec 27, 2017 •

edited

Loading