KAFKA-7798: Expose embedded clientIds by guozhangwang · Pull Request #6107 · apache/kafka

guozhangwang · 2019-01-09T02:29:07Z

Add consumer / restoreConsumer / producer(s) / admin client ids via ThreadMetadata; for producerIds, if EOS is turned on add the list of task-producer-ids, otherwise it is a singleton of thread-producer-id.
Consolidate the logic of generating clientIds from thread names and client id into StreamThread.

Committer Checklist (excluded from commit message)

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

guozhangwang · 2019-01-09T03:10:46Z

abbccdda · 2019-01-09T06:59:06Z


        final StreamThread.StreamsMetricsThreadImpl streamsMetrics = new StreamThread.StreamsMetricsThreadImpl(metrics, "");
        final StreamThread thread = new StreamThread(
-                mockTime,


Why we change the style here?

I just did a copy-paste of the code and this is intellij auto formatting. Current style seems right to me.

bbejeck

Thanks @guozhangwang for the patch. Overall looks good to me, but I'm getting some test failures locally on StreamThreadTest.

bbejeck

Ignore this comment, I posted twice by accident.

guozhangwang · 2019-01-12T01:24:46Z

@bbejeck That's due to a recent commit that caused conflicts, pushed another commit to resolve it.

abbccdda

LGTM overall. Just left some comments

abbccdda · 2019-01-13T04:37:00Z

+        return threadClientId + "-restore-consumer";
+    }
+
+    public static String getSharedAdminClientId(final String clientId) {


Could we have a comment on explaining what shared means in this context?

abbccdda · 2019-01-13T04:43:51Z

+        return this;
+    }
+
+    private void updateThreadMetadata(final Map<TaskId, StreamTask> activeTasks,


Could we keep the format of updateThreadMetadata to avoid unnecessary changes on L1221?

I cannot completely follow since L1221 did not change, could you elaborate?

I mean, if we keep parameters in one line:
private void updateThreadMetadata(final Map<TaskId, StreamTask> activeTasks, final Map<TaskId, StandbyTask> standbyTasks) then the changes shall be reduced.

The line is quite long, thus we should reformat it to make the code more readable. I am +1 on this change -- also, I don't think we should optimize for fewer line changes :)

abbccdda · 2019-01-13T04:47:11Z

+    // package-private for testing only
+    StreamThread updateThreadMetadata(final String adminClientId) {
+
+        threadMetadata = new ThreadMetadata(


Since this constructor is only called once, could we just reuse the constructor augmented:

public ThreadMetadata(final String threadName, final String threadState, final String mainConsumerClientId, final String restoreConsumerClientId, final Set<String> producerClientIds, final String adminClientId, final Set<TaskMetadata> activeTasks, final Set<TaskMetadata> standbyTasks)

and set active & standby tasks with empty set? I figure this could reduce code redundancy.

abbccdda · 2019-01-13T04:48:33Z

        if (globalStreamThread != null) {
            result.putAll(globalStreamThread.consumerMetrics());
        }
+        // own streams metrics


Own could sound like a verb. Maybe self streams metrics will be better?

…bed-client-context

guozhangwang · 2019-01-18T05:18:51Z

@mjsax could you give another look on this?

dguy

Thanks @guozhangwang left a few minor comments

dguy · 2019-01-18T17:59:13Z

    @Deprecated
-    public Map<String, Object> getConsumerConfigs(final String groupId,
-                                                  final String clientId) {
+    public Map<String, Object> getConsumerConfigs(final String groupId, final String clientId) {


don't we usually split params over multiple lines?

It's a subjective question. I personally prefer to have one parameter per line. :)

Yeah I personally prefer only doing that for 3+ parameters and they cannot fit in 150 chars which is what a normal laptop screen would fit.

dguy · 2019-01-18T17:59:20Z

    @SuppressWarnings("WeakerAccess")
-    public Map<String, Object> getMainConsumerConfigs(final String groupId,
-                                                      final String clientId) {
+    public Map<String, Object> getMainConsumerConfigs(final String groupId, final String clientId) {


dguy · 2019-01-18T18:02:01Z

+        return threadClientId + "-producer";
+    }
+
+    private static String getThreadConsumerClientId(final String threadClientId) {


nit: I guess this and the next one could be getConsumerClientId and getRestoreConsumerId as there is only ever one of each?

dguy · 2019-01-18T18:06:38Z

        final String clientId = "client";
        final Map<String, Object> returnedProps = streamsConfig.getProducerConfigs(clientId);
-        assertEquals(returnedProps.get(ProducerConfig.CLIENT_ID_CONFIG), clientId + "-producer");
+        assertEquals(returnedProps.get(ProducerConfig.CLIENT_ID_CONFIG), clientId);


shouldn't the arguments be the other way around? i.e., assertEquals(expected, actual) ?
better still use assertThat() :-P

dguy · 2019-01-18T18:07:10Z

        final String clientId = "client";
        final Map<String, Object> returnedProps = streamsConfig.getMainConsumerConfigs(groupId, clientId);
-        assertEquals(returnedProps.get(ConsumerConfig.CLIENT_ID_CONFIG), clientId + "-consumer");
+        assertEquals(returnedProps.get(ConsumerConfig.CLIENT_ID_CONFIG), clientId);


ditto here. I know it was wrong before, but would be good to correct.

mjsax

Couple of nits.

mjsax · 2019-01-18T17:56:40Z

        }

-        // add client id with stream client id prefix, and group id
+        // this is a hack to work around StreamsConfig constructor inside StreamPartitionAssignor to avoid casting


Don't understand this comment? Can you elaborate?

Inside StreamPartitionAssignor we wrap the passed-in props map into a InternalStreamsConfig in order to read its config values without casting. But InternalStreamsConfig extends StreamsConfig requires application id as a must-have, so we need to pass in this value to consumer config prop values as well.

Ack. Makes sense.

However, this is not really clear from the comment. Maybe rephrase to avoid confusion?

mjsax · 2019-01-18T18:01:03Z

            logContext,
-            assignmentErrorCode);
+            assignmentErrorCode)
+            .updateThreadMetadata(getSharedAdminClientId(clientId));


Why do we not pass the admin client-id into the constructor?

I've thought about passing in the client-id, it is a bit overkill in addition to the thread-client-id just to set the shared admin client inside the constructor. I felt the current way is less intrusive but if you feel strong the other way we can discuss.

Because the shared admit client id does never change, it seems weird to have a setter that allow to update is at any time. It seems cleaner to pass it in the constructor and make it immutable this way.

mjsax · 2019-01-18T18:02:31Z

+        return this;
+    }
+
+    private void updateThreadMetadata(final Map<TaskId, StreamTask> activeTasks,


The line is quite long, thus we should reformat it to make the code more readable. I am +1 on this change -- also, I don't think we should optimize for fewer line changes :)

mjsax · 2019-01-18T18:06:56Z

    @Override
    public int hashCode() {
-        return Objects.hash(threadName, threadState, activeTasks, standbyTasks);
+        return Objects.hash(threadName, threadState, activeTasks, standbyTasks, mainConsumerClientId, restoreConsumerClientId, producerClientIds, adminClientId);


Nit: Can we break this long line into one parameter per line?

mjsax · 2019-01-18T18:08:25Z

    public void before() {
        props = new Properties();
        props.put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
+        props.put(StreamsConfig.CLIENT_ID_CONFIG, "appId");


Nit: why reuse appId -- if there is a bug, and we set APPLICATION_ID_CONFIG by mistake, it would not be detected. Maybe better to set a different id.

bbejeck

LGTM modulo comments from @dguy

…bed-client-context

guozhangwang · 2019-01-18T20:01:55Z

Updated per your comments @mjsax @dguy

…bed-client-context

vvcephei · 2019-01-30T22:04:59Z

            result.putAll(thread.consumerMetrics());
+            // admin client is shared, so we can actually move it
+            // to result.putAll(adminClient.metrics()).
+            // we did it intentionally just for flexibility.


why not go ahead and do this now?

mjsax · 2019-02-02T05:27:03Z

@guozhangwang I am merging this to cut the release branch. Please address the left over comment in a follow up PR. Thanks.

* AK/trunk: fix typo (apache#5150) MINOR: Reduce replica.fetch.backoff.ms in ReassignPartitionsClusterTest (apache#5887) KAFKA-7766: Fail fast PR builds (apache#6059) KAFKA-7798: Expose embedded clientIds (apache#6107) KAFKA-7641; Introduce "group.max.size" config to limit group sizes (apache#6163) KAFKA-7433; Introduce broker options in TopicCommand to use AdminClient (KIP-377) MINOR: Fix some field definitions for ListOffsetReponse (apache#6214) KAFKA-7873; Always seek to beginning in KafkaBasedLog (apache#6203) KAFKA-7719: Improve fairness in SocketServer processors (KIP-402) (apache#6022) MINOR: fix checkstyle suppressions for generated RPC code to work on Windows KAFKA-7859: Use automatic RPC generation in LeaveGroups (apache#6188) KAFKA-7652: Part II; Add single-point query for SessionStore and use for flushing / getter (apache#6161) KAFKA-3522: Add RocksDBTimestampedStore (apache#6149) KAFKA-3522: Replace RecordConverter with TimestampedBytesStore (apache#6204)

Reviewers: Damian Guy <damian@confluent.io>, John Roesler <john@confluent.io>, Boyang Chen <bchen11@outlook.com>, Matthias J. Sax <matthias@confluent.io>, Bill Bejeck <bill@confluent.io>

guozhangwang added 4 commits January 8, 2019 18:06

add client ids

4200501

add back

77b6b9f

add back unit test

88b8bf9

add unit test

cf31312

guozhangwang changed the title ~~KAFKA-7798: expose embedded client context~~ KAFKA-7798: Expose embedded clientIds Jan 9, 2019

abbccdda reviewed Jan 9, 2019

View reviewed changes

bbejeck reviewed Jan 11, 2019

View reviewed changes

rebase from trunk

9df04b3

abbccdda reviewed Jan 13, 2019

View reviewed changes

mjsax added the streams label Jan 14, 2019

guozhangwang added 2 commits January 15, 2019 14:10

Merge branch 'trunk' of https://github.com/apache/kafka into K7798-em…

568e73e

…bed-client-context

github comments

301abc5

dguy reviewed Jan 18, 2019

View reviewed changes

mjsax reviewed Jan 18, 2019

View reviewed changes

bbejeck approved these changes Jan 18, 2019

View reviewed changes

guozhangwang added 2 commits January 18, 2019 11:37

Merge branch 'trunk' of https://github.com/apache/kafka into K7798-em…

4d5be48

…bed-client-context

github comments

b35dc3b

Merge branch 'trunk' of https://github.com/apache/kafka into K7798-em…

65e8c14

…bed-client-context

vvcephei reviewed Jan 30, 2019

View reviewed changes

mjsax merged commit 9dc76f8 into apache:trunk Feb 2, 2019

guozhangwang deleted the K7798-embed-client-context branch April 25, 2020 00:05

Conversation

guozhangwang commented Jan 9, 2019

Committer Checklist (excluded from commit message)

Uh oh!

guozhangwang commented Jan 9, 2019

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bbejeck left a comment • 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!

bbejeck left a comment • 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!

guozhangwang commented Jan 12, 2019

Uh oh!

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

abbccdda Jan 16, 2019 • 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!

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!

guozhangwang commented Jan 18, 2019

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!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

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

Choose a reason for hiding this comment

bbejeck left a comment •

edited

Loading

bbejeck left a comment •

edited

Loading

abbccdda Jan 16, 2019 •

edited

Loading