KAFKA-5226: Fixes issue where adding topics matching a regex #3157

bbejeck · 2017-05-26T21:47:59Z

subscribed stream may not be detected by all followers until
onJoinComplete returns.

subscribed stream may not be detected by all followers until onJoinComplete returns.

bbejeck · 2017-05-26T21:52:37Z

This PR address situation where adding topics matching a regex subscribed stream may not be detected by all followers in StreamPartitionAssignor.subscribe but are picked up when
ConsumerCoordinator.onJoinComplete returns.

ping @mjsax @dguy @guozhangwang for review

asfbot · 2017-05-26T22:44:56Z

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

asfbot · 2017-05-26T23:10:37Z

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

bbejeck · 2017-05-26T23:26:23Z

The test failure is kafka.admin.DescribeConsumerGroupTest.testDescribeExistingGroupWithNoMembersWithNewConsumer so I think this is unrelated to streams and this PR

bbejeck · 2017-05-26T23:26:36Z

retest this please

asfbot · 2017-05-27T00:26:44Z

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

asfbot · 2017-05-27T00:32:21Z

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

guozhangwang · 2017-05-30T06:36:35Z

Good find!

About the fix itself, in ConsumerCoordinator we already have the logic for handling unknown topics coming from the leader's assignment:

// check if the assignment contains some topics that were not in the original
        // subscription, if yes we will obey what leader has decided and add these topics
        // into the subscriptions as long as they still match the subscribed pattern
        //
        // TODO this part of the logic should be removed once we allow regex on leader assign
        Set<String> addedTopics = new HashSet<>();
        for (TopicPartition tp : subscriptions.assignedPartitions()) {
            if (!joinedSubscription.contains(tp.topic()))
                addedTopics.add(tp.topic());
        }

        if (!addedTopics.isEmpty()) {
            Set<String> newSubscription = new HashSet<>(subscriptions.subscription());
            Set<String> newJoinedSubscription = new HashSet<>(joinedSubscription);
            newSubscription.addAll(addedTopics);
            newJoinedSubscription.addAll(addedTopics);

            this.subscriptions.subscribeFromPattern(newSubscription);
            this.joinedSubscription = newJoinedSubscription;
        }

        // update the metadata and enforce a refresh to make sure the fetcher can start
        // fetching data in the next iteration
        this.metadata.setTopics(subscriptions.groupSubscription());
        client.ensureFreshMetadata();

I think a better way to fix it is to modify StreamPartitionAssignor#onAssignment which is called after the above code and hence when it is called the metadata should have been refreshed. And in that call we can trigger streamThread.builder.updateSubscriptions().

Also if we do that in onAssignment, maybe we do not need to call updateSubscription in StreamPartitionAssignor#subscription() any more?

guozhangwang · 2017-05-30T06:42:56Z

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

+
+        try {
+            streamsLeader.start();
+            Thread.sleep(1000);


I'd suggest avoid Thread.sleep() as much as possible in unit test as it will introduce timing-dependent flakiness. If there is any condition that we need to wait on consider using TestUtils.waitForCondition.

Also this test seem not always testing the race condition that can cause the issue; I'd suggest make it really "modular" by adding a test for StreamThread, in which we do not need to start the thread but just mocking its embedded StreamPartitionAssignor#onAssignment() which gives it some assigned partition whose topic is not unknown, and then call its onPartitionAssigned callback to check if the topology builder has updated its topology.

I'll remove the first two, but ok to leave in the one on line 352? I want to space the the creation of topics.

I'm actually thinking more about making the unit test really "unit" by mocking stream partition assignor to always return unknown topic partitions; the current test case seems to be more "integration test"?

Yes it is, I see your point now about the test, I'll make the changes.

bbejeck · 2017-05-30T15:56:52Z

@guozhangwang thanks for the review. I'll move the fix to the StreamPartitionAssignor#onAssignment .

From what I can see we'll need to leave the to call to TopologyBuilder.updateSubscription in StreamPartitionAssignor#subscription() as it updates the leader's topology used in theStreamPartitionAssignor.assign method.

guozhangwang · 2017-05-30T19:13:54Z

From what I can see we'll need to leave the to call to TopologyBuilder.updateSubscription in StreamPartitionAssignor#subscription() as it updates the leader's topology used in theStreamPartitionAssignor.assign method.

makes sense.

bbejeck · 2017-05-30T22:45:04Z

@guozhangwang updates per comments.

Put SubscriptionUpdates back as inner class of StreamPartitionAssignor
Removed integration test and added unit test
Put the check for updated topics in the StreamPartitionAssignor#onAssignment

mjsax

Just some nits. Overall LGTM.

mjsax · 2017-05-30T23:32:28Z

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

+
+
+    private void updateSubscribedTopics(Assignment assignment) {
+        if (streamThread != null && streamThread.builder.sourceTopicPattern() != null) {


How could streamThread be null -- null would indicate a bug and we might want to fail hard on that instead of masking it with this check?

you're right it can't, I put that in for unit tests, I'll remove and update the tests

mjsax · 2017-05-30T23:32:37Z

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

+    }
+
+
+    private void updateSubscribedTopics(Assignment assignment) {


nit: add final
also in L616

mjsax · 2017-05-30T23:34:51Z

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

+            for (TopicPartition topicPartition : assignment.partitions()) {
+                assignedTopics.add(topicPartition.topic());
+            }
+            if (!assignedTopics.isEmpty() && !streamThread.builder.subscriptionUpdates().getUpdates().containsAll(assignedTopics)) {


Would containsAll not return true if assignedTopics is empty? (ie, can we remove the first check?)

Good catch, containsAll returns if assignedTopics is empty, removing first check.

asfbot · 2017-05-31T00:00:46Z

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

asfbot · 2017-05-31T00:18:23Z

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

bbejeck · 2017-05-31T00:29:18Z

@mjsax removed null check for streamThread and first check for empty assignedTopic

cc\ @guozhangwang

asfbot · 2017-05-31T01:31:56Z

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

asfbot · 2017-05-31T01:35:15Z

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

guozhangwang

Thanks for the update! LGTM overall, just a few minor comments.

guozhangwang · 2017-05-31T06:28:41Z

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

+    }
+
+
+    private void updateSubscribedTopics(final Assignment assignment) {


nit: could we change this to maybeUpdateSubscribedTopics(Set<String> topics) with the check on streamThread.builder.sourceTopicPattern() != null within the function so that both subscription() and onAssignment could consolidate on this function?

Yes. This will cause a rebuild of the topology in those cases where all topics are already known. But thinking about it some more, IMHO it's worth the tradeoff for simplifying the logic.

guozhangwang · 2017-05-31T06:30:26Z

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

@@ -104,6 +106,12 @@
    private final MockClientSupplier mockClientSupplier = new MockClientSupplier();
    private final TopologyBuilder builder = new TopologyBuilder();
    private final StreamsConfig config = new StreamsConfig(configProps());
+    private final StreamThread mockStreamThread = new StreamThread(builder, config,


Why modify this test class to let partitionAssignor set this mock stream thread?

There are some tests using onAssign which were getting NPEs. Since in practice the embedded streamThread should never be null, I opted to update the test.

guozhangwang · 2017-05-31T06:37:13Z

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

+        assignment = new PartitionAssignor.Assignment(topicPartitions, info.encode());
+        partitionAssignor.onAssignment(assignment);
+
+        assertTrue(nodeToSourceTopics.get("source").size() == 3);


Is the third step (i.e. from 2 topics to 3 topics) intended for additional test coverage?

Yes I wanted to exercise the functionality more than once.

bbejeck · 2017-05-31T15:23:36Z

@guozhangwang thanks for the additional pass, updates per comments.

asfbot · 2017-05-31T16:40:44Z

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

asfbot · 2017-05-31T18:00:09Z

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

bbejeck · 2017-05-31T18:00:57Z

retest this please

asfbot · 2017-05-31T19:36:15Z

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

asfbot · 2017-05-31T19:39:59Z

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

bbejeck · 2017-05-31T20:18:03Z

previous failure was unrelated to this PR

mjsax

LGTM. (beside the reflection comment, but this should not block the PR).

mjsax · 2017-05-31T23:57:50Z

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

+        final Field
+            nodeToSourceTopicsField =
+            topologyBuilder.getClass().getDeclaredField("nodeToSourceTopics");
+        nodeToSourceTopicsField.setAccessible(true);


Arg. Reflections. Seems like our code is not well structure if we need this. Can you file a Jira to do some follow up cleanup (or if this does not make it into 0.11, we might to the cleanup first).

… discovered.

bbejeck · 2017-06-01T00:20:12Z

@guozhangwang @mjsax updated per comments and discussion on slack.

Specifically, now we'll only rebuild the topology under two conditions

The topics passed into StreamPartitionAssignor.subscription don't exactly match what the current TopologyBuilder.subscriptionUpdates contains
Topics passed into StreamPartitionAssignor.onAssign have newly discovered topics from ConsumerCoordinator.onJoinComplete

asfbot · 2017-06-01T01:25:59Z

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

asfbot · 2017-06-01T01:26:19Z

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

guozhangwang · 2017-06-01T02:36:39Z

LGTM! Merged to trunk and 0.11.0.

subscribed stream may not be detected by all followers until onJoinComplete returns. Author: Bill Bejeck <bill@confluent.io> Reviewers: Matthias J. Sax <matthias@confluent.io>, Guozhang Wang <wangguoz@gmail.com> Closes #3157 from bbejeck/KAFKA-5226_null_pointer_source_node_deserialize (cherry picked from commit 6360e04) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

KAFKA-5226: Fixes issue where adding topics matching a regex

ed3c360

subscribed stream may not be detected by all followers until onJoinComplete returns.

guozhangwang reviewed May 30, 2017

View reviewed changes

KAFKA-5226: Updates per comments

b2b6436

mjsax reviewed May 30, 2017

View reviewed changes

KAFKA-5226: Updates per comments

b55af72

guozhangwang reviewed May 31, 2017

View reviewed changes

KAFKA-5226: Updates per comments

a28c2f6

mjsax approved these changes Jun 1, 2017

View reviewed changes

KAFKA-5226: Only update topics (hence rebuild topology) if new topics…

cd4eeee

… discovered.

asfgit closed this in 6360e04 Jun 1, 2017

bbejeck deleted the KAFKA-5226_null_pointer_source_node_deserialize branch July 10, 2024 13:55



		private void updateSubscribedTopics(Assignment assignment) {
		if (streamThread != null && streamThread.builder.sourceTopicPattern() != null) {

		}


		private void updateSubscribedTopics(final Assignment assignment) {

KAFKA-5226: Fixes issue where adding topics matching a regex #3157

KAFKA-5226: Fixes issue where adding topics matching a regex #3157

Conversation

bbejeck commented May 26, 2017

bbejeck commented May 26, 2017

asfbot commented May 26, 2017

asfbot commented May 26, 2017

bbejeck commented May 26, 2017

bbejeck commented May 26, 2017

asfbot commented May 27, 2017

asfbot commented May 27, 2017

guozhangwang commented May 30, 2017

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 commented May 30, 2017

guozhangwang commented May 30, 2017

bbejeck commented May 30, 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

asfbot commented May 31, 2017

asfbot commented May 31, 2017

bbejeck commented May 31, 2017

asfbot commented May 31, 2017

asfbot commented May 31, 2017

guozhangwang 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

bbejeck commented May 31, 2017

asfbot commented May 31, 2017

asfbot commented May 31, 2017

bbejeck commented May 31, 2017

asfbot commented May 31, 2017

asfbot commented May 31, 2017

bbejeck commented May 31, 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

bbejeck commented Jun 1, 2017

asfbot commented Jun 1, 2017

asfbot commented Jun 1, 2017

guozhangwang commented Jun 1, 2017