Kafka 4060 remove zk client dependency in kafka streams followup re-branched from trunk by hjafarpour · Pull Request #2389 · apache/kafka

hjafarpour · 2017-01-17T22:27:14Z

Re-branched the trunk and applied the changes to the new branch to simplify commit log.

…plify the commit log..

asfbot · 2017-01-17T23:15:39Z

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

asfbot · 2017-01-17T23:16:30Z

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

asfbot · 2017-01-17T23:17:43Z

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

asfbot · 2017-01-18T01:36:46Z

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

asfbot · 2017-01-18T01:47:49Z

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

asfbot · 2017-01-18T02:10:43Z

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

ijuma · 2017-01-18T09:22:44Z

                } else {
-                    throw new StreamsException("Inconsistent response received from broker " + brokerId +
-                            ", expected correlation id " + clientRequest.correlationId() + ", but received " +
+                    throw new StreamsException("Inconsistent response received from the broker, expected correlation id " + clientRequest.correlationId() + ", but received " +


Out of curiosity, why do we remove the broker id from the message?

The broker id is not available and I didn't want to call getBrokerId.

ClientRequest.destination is the broker id.

enothereska · 2017-01-18T09:50:16Z

This PR removes the delete topic, and throws an exception is the number of partitions has changed. However, the number of partitions always changes in some cases, due to the topic auto-create issue. As I understand it, with this PR an app will fail. How do we address that? @hjafarpour @guozhangwang

dguy

Left some comments. Also, AFAICT StreamsKafkaClient doesn't have any tests - this should be fixed

dguy · 2017-01-18T09:46:54Z

            if (existingTopicNamesPartitions.get(topic.name()) != null) {
                if (existingTopicNamesPartitions.get(topic.name()) != topicsPartitionsMap.get(topic)) {
-                    deleteTopics.put(topic, topicsPartitionsMap.get(topic));
+                    throw new StreamsException("Internal topic with invalid partitons. Use 'kafka.tools.StreamsResetter' tool to clean up invalid topics before proceesing.");


typo: - proceesing -> processing

dguy · 2017-01-18T09:50:04Z

+        final ClientResponse clientResponse = sendRequest(clientRequest);
+
        if (!(clientResponse.responseBody() instanceof CreateTopicsResponse)) {
            throw new StreamsException("Inconsistent response type for internal topic creation request. Expected CreateTopicsResponse but received " + clientResponse.responseBody().getClass().getName());


This probably needs to check clientResponse.hasResponse() otherwise it could throw NullPointerException

Added the check.

dguy · 2017-01-18T09:51:13Z

+        final ClientRequest clientRequest = kafkaClient.newClientRequest(getBrokerId(), new MetadataRequest.Builder(Arrays.asList(topic)), Time.SYSTEM.milliseconds(), true, null);
+        final ClientResponse clientResponse = sendRequest(clientRequest);
        if (!(clientResponse.responseBody() instanceof MetadataResponse)) {
            throw new StreamsException("Inconsistent response type for internal topic metadata request. Expected MetadataResponse but received " + clientResponse.responseBody().getClass().getName());


Same as above: need to check clientResponse.hasResponse()

Added the check.

dguy · 2017-01-18T09:52:44Z

+    public Collection<MetadataResponse.TopicMetadata> fetchTopicsMetadata() {
+
+        final ClientRequest clientRequest = kafkaClient.newClientRequest(getBrokerId(), new MetadataRequest.Builder(null), Time.SYSTEM.milliseconds(), true, null);
+        final ClientResponse clientResponse = sendRequest(clientRequest);


Added the check.

enothereska · 2017-01-18T10:04:13Z

Ok, that seems to be in a different PR: #2379. Got it, thanks.

enothereska · 2017-01-18T09:48:05Z

-     * partitions delete them and create new ones with correct number of partitons along with the non existing topics.
+     * Prepares a given internal topic.
+     * If the topic does not exist creates a new topic.
+     * If the topic with the correct number of partitions exists ignores it.


Add "...exists"

guozhangwang · 2017-01-18T19:57:18Z

@hjafarpour Could you address the above comments?

asfbot · 2017-01-18T22:29:22Z

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

asfbot · 2017-01-18T22:30:19Z

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

asfbot · 2017-01-18T23:03:49Z

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

asfbot · 2017-01-18T23:25:43Z

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

asfbot · 2017-01-18T23:25:58Z

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

asfbot · 2017-01-18T23:29:42Z

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

…e already exist with unexpected number of partitions Re-branched the trunk and applied the changes to the new branch to simplify commit log. Author: Hojjat Jafarpour <hojjat@Hojjat-Jafarpours-MBP.local> Reviewers: Ismael Juma, Damian Guy, Eno Thereska, Guozhang Wang Closes #2389 from hjafarpour/KAFKA-4060-Remove-ZkClient-dependency-in-Kafka-Streams-followup-from-trunk Address Ismael's comments upon merging (cherry picked from commit 8e2cbae) Signed-off-by: Guozhang Wang <wangguoz@gmail.com>

guozhangwang · 2017-01-18T23:55:11Z

Addressed @ijuma 's comments upon merging to trunk and 0.10.2.

…e already exist with unexpected number of partitions Re-branched the trunk and applied the changes to the new branch to simplify commit log. Author: Hojjat Jafarpour <hojjat@Hojjat-Jafarpours-MBP.local> Reviewers: Ismael Juma, Damian Guy, Eno Thereska, Guozhang Wang Closes apache#2389 from hjafarpour/KAFKA-4060-Remove-ZkClient-dependency-in-Kafka-Streams-followup-from-trunk Address Ismael's comments upon merging

Hojjat Jafarpour added 2 commits January 17, 2017 14:23

Rebranched the trunk and applied the changes to the new branch to sim…

8b8752e

…plify the commit log..

Removed formatting changes from the previous commit.

e914532

Minor format corrections.

4f9ed25

ijuma reviewed Jan 18, 2017

View reviewed changes

dguy requested changes Jan 18, 2017

View reviewed changes

enothereska reviewed Jan 18, 2017

View reviewed changes

Minor changes based on review feedback.

b649e81

Typo correction.

3ec69a2

asfgit closed this in 8e2cbae Jan 18, 2017

Conversation

hjafarpour 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!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 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!

enothereska commented Jan 18, 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!

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!

enothereska commented Jan 18, 2017

Uh oh!

Choose a reason for hiding this comment

Uh oh!

guozhangwang commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

asfbot commented Jan 18, 2017

Uh oh!

guozhangwang commented Jan 18, 2017

Uh oh!

Reviewers

Assignees

Labels

Projects

Milestone

Development

Uh oh!

6 participants