KAFKA-4706: Unify StreamsKafkaClient instances #2631

sharad-develop · 2017-03-03T02:08:08Z

KAFKA:4706 - Unify StreamsKafkaClient instances

asfbot · 2017-03-03T02:10:14Z

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

asfbot · 2017-03-03T02:10:27Z

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

asfbot · 2017-03-03T02:10:56Z

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

asfbot · 2017-03-03T03:04:09Z

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

asfbot · 2017-03-03T03:04:35Z

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

asfbot · 2017-03-03T03:05:23Z

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

mjsax

Thanks for the PR @sharad-develop

Please wait for additional comments from @enothereska or @dguy before updating the PR (to avoid going forth and back :))

\cc @enothereska @dguy for second review

mjsax · 2017-03-03T03:54:41Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-                                          time,
-                                          streamsMetadataState,
-                                          cacheSizeBytes);
+                    config,


Nit: move builder one line down to get aligned indention.

mjsax · 2017-03-03T03:59:15Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

-        } catch (final IOException e) {
-            log.warn("Could not close StreamKafkaClient.", e);
-        }
+        streamsKafkaClient.checkBrokerCompatibility();


KafkaStreams should close StreamsKafkaClient in it's own close() method. Also, InternalTopicManager should not close StreamKafkaClient anymore.

mjsax · 2017-03-03T03:59:43Z

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

@@ -207,7 +207,7 @@ public void configure(Map<String, ?> configs) {
        }

        internalTopicManager = new InternalTopicManager(
-                new StreamsKafkaClient(this.streamThread.config),
+                this.streamThread.getStreamsKafkaClient(),


Nit: remove this (we try to avoid this when possible)

mjsax · 2017-03-03T04:00:06Z

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

@@ -1103,6 +1104,22 @@ public String toString(final String indent) {
    }

    /**
+     *


Add JavaDocs (if we keep the setter)

mjsax · 2017-03-03T04:00:14Z

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

+    }
+
+    /**
+     *


Add JavaDocs (if we keep the getter)

mjsax · 2017-03-03T04:01:54Z

streams/src/main/java/org/apache/kafka/streams/KafkaStreams.java

+                    time,
+                    streamsMetadataState,
+                    cacheSizeBytes);
+            threads[i].setStreamsKafkaClient(streamsKafkaClient);


I would add this to the constructor instead of a setter.
\cc @enothereska @dguy WDYT?

I initially started with a constructor approach. But the checkstyle was making the build fail as the constructor exceeded 10 arguments.

I think constructor is a better approach , but not sure if there is a way around checkstyle.

I guess you can remove parameter applicationId because it is contained in the config anyway.

Ok. Should I go ahead with the constructor approach or are we still deliberating on alternative approaches which are getting discussed ?

I guess we are still discussing...

mjsax · 2017-03-03T04:02:34Z

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

@@ -203,6 +203,7 @@ private synchronized void setStateWhenNotInPendingShutdown(final State newState)
    final StateDirectory stateDirectory;
    private String originalReset;
    private StreamPartitionAssignor partitionAssignor = null;
+    private StreamsKafkaClient streamsKafkaClient;


Can be final is we set it via constructor instead of setter.

mjsax · 2017-03-03T04:10:21Z

@sharad-develop Can you please update the PR title to KAFKA-4706: Unify StreamsKafkaClient instances -- this allows the bot to link this PR to the JIRA automatically. Thx.

asfbot · 2017-03-03T06:15:10Z

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

asfbot · 2017-03-03T06:15:20Z

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

asfbot · 2017-03-03T06:15:26Z

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

dguy · 2017-03-03T08:14:23Z

There is a bunch of streams test failures

dguy

I'm thinking we can extract an interface from StreamsKafkaClient - i'm not 100% sure what will be called yet (maybe AdminClient ?). Then we add a method to KafkaClientSupplier, i.e., AdminClient getAdminClient(final StreamsConfig config)

This would mean that creation of all Kafka related clients is behind the same interface. Which means that it will eventually help us with being able to decouple our integration tests from real brokers.
It also means that we don't need to add any setters or constructor params to the StreamThread as it already accepts a KafkaClientSupplier as an argument. If we do only really want one instance of the StreamsKafkaClient then this can be done easily enough in the DefaultKafkaClientSupplier.

@mjsax @enothereska @guozhangwang - thoughts?

mjsax · 2017-03-03T18:17:56Z

AdminClient is already used by core (https://issues.apache.org/jira/browse/KAFKA-3265) - so not sure if the same name is a good choice. This raised the question, if we (at some point) could get rid of StreamKafkaClient and use AdminClient only?

I like the KafkaClientSupplier idea, however, this will require a KIP.

guozhangwang · 2017-03-06T17:57:44Z

@dguy I like your idea, and I think the name of AdminClient is fine: for example we can add another method named getAdmin in KafkaClientSupplier which can then be used in both KafkaStreams then passed into each of the thread's StreamsPartitionAssignor.

guozhangwang · 2017-03-06T19:12:42Z

BTW I think we can also piggy-back this with KIP-117 which will introduce an admin client in common package that is can be generally used. Today we have different impls of such client in streams and connect, and that KIP is aimed to condense all these clients.

mjsax · 2017-03-06T23:12:01Z

@guozhangwang Do you suggest to delay this PR until KIP-117 is done? This might take some time, as the KIP is not even voted yet.

guozhangwang · 2017-03-06T23:25:22Z

That's a good point: let's not wait for KIP-117 then, and we can do the merge in a separate JIRA after that is done.

mjsax · 2017-03-06T23:42:50Z

Sounds good to me. I did create a JIRA for this: https://issues.apache.org/jira/browse/KAFKA-4857

@sharad-develop Can you follow up with this PR and address the comments. Thanks.

guozhangwang · 2018-01-02T18:42:17Z

Closing this PR for now.

KAFKA:4706 - Unify StreamsKafkaClient instances

906eeae

KAFKA:4706 - Unify StreamsKafkaClient instances

a30b719

KAFKA:4706 - Unify StreamsKafkaClient instances

b576b36

mjsax requested changes Mar 3, 2017

View reviewed changes

dguy reviewed Mar 3, 2017

View reviewed changes

sharad-develop changed the title ~~KAFKA:4706 - Unify StreamsKafkaClient instances~~ KAFKA-4706: Unify StreamsKafkaClient instances Mar 3, 2017

guozhangwang closed this Jan 2, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-4706: Unify StreamsKafkaClient instances #2631

KAFKA-4706: Unify StreamsKafkaClient instances #2631

sharad-develop commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

mjsax left a comment

mjsax Mar 3, 2017

mjsax Mar 3, 2017

mjsax Mar 3, 2017

mjsax Mar 3, 2017

mjsax Mar 3, 2017

mjsax Mar 3, 2017

sharad-develop Mar 3, 2017

sharad-develop Mar 3, 2017

mjsax Mar 3, 2017

sharad-develop Mar 3, 2017

mjsax Mar 4, 2017

mjsax Mar 3, 2017

mjsax commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

dguy commented Mar 3, 2017

dguy left a comment

mjsax commented Mar 3, 2017

guozhangwang commented Mar 6, 2017

guozhangwang commented Mar 6, 2017

mjsax commented Mar 6, 2017

guozhangwang commented Mar 6, 2017

mjsax commented Mar 6, 2017

guozhangwang commented Jan 2, 2018

@@ @@ -1103,6 +1104,22 @@ public String toString(final String indent) { @@
                   }
                   /**
+                   *

KAFKA-4706: Unify StreamsKafkaClient instances #2631

KAFKA-4706: Unify StreamsKafkaClient instances #2631

Conversation

sharad-develop commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 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

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

mjsax commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

asfbot commented Mar 3, 2017

dguy commented Mar 3, 2017

dguy left a comment

Choose a reason for hiding this comment

mjsax commented Mar 3, 2017

guozhangwang commented Mar 6, 2017

guozhangwang commented Mar 6, 2017

mjsax commented Mar 6, 2017

guozhangwang commented Mar 6, 2017

mjsax commented Mar 6, 2017

guozhangwang commented Jan 2, 2018