[GitHub] maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify clientId to groupId in flink-connector-kafka-0.8
maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify
clientId to groupId in flink-connector-kafka-0.8
URL: https://github.com/apache/flink/pull/5304#discussion_r209480301
##
File path:
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
##
@@ -123,6 +124,7 @@ public SimpleConsumerThread(
this.fetchSize = getInt(config, "fetch.message.max.bytes",
1048576);
this.bufferSize = getInt(config, "socket.receive.buffer.bytes",
65536);
this.reconnectLimit = getInt(config,
"flink.simple-consumer-reconnectLimit", 3);
+ this.clientId = config.getProperty("group.id",
"flink-kafka-consumer-legacy-" + broker.id());
Review comment:
The way that @azagrebin is suggesting is really my initially suggested, but
` "flink-kafka-consumer-legacy-" + broker.id() ` has not been in effect.
I'm a little confused now, how do I set it up better.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify clientId to groupId in flink-connector-kafka-0.8
maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify
clientId to groupId in flink-connector-kafka-0.8
URL: https://github.com/apache/flink/pull/5304#discussion_r209138484
##
File path:
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
##
@@ -123,6 +124,7 @@ public SimpleConsumerThread(
this.fetchSize = getInt(config, "fetch.message.max.bytes",
1048576);
this.bufferSize = getInt(config, "socket.receive.buffer.bytes",
65536);
this.reconnectLimit = getInt(config,
"flink.simple-consumer-reconnectLimit", 3);
+ this.clientId = config.getProperty("group.id",
"flink-kafka-consumer-legacy-" + broker.id());
Review comment:
Thank you for the review @azagrebin and @aljoscha.
I looked at the kafka document again, for `kafka consumer`, group.id is a
required property.
- for old consumer, the default value set "group.id value".
- for new consumer, the default value set "".
so, i agree with @aljoscha, clear and concise.
```
this.clientId = config.getProperty("client.id",
"flink-kafka-consumer-legacy-" + broker.id());
```
If ok, I'll resubmit it again, thanks.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify clientId to groupId in flink-connector-kafka-0.8
maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify
clientId to groupId in flink-connector-kafka-0.8
URL: https://github.com/apache/flink/pull/5304#discussion_r208790526
##
File path:
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
##
@@ -123,6 +124,7 @@ public SimpleConsumerThread(
this.fetchSize = getInt(config, "fetch.message.max.bytes",
1048576);
this.bufferSize = getInt(config, "socket.receive.buffer.bytes",
65536);
this.reconnectLimit = getInt(config,
"flink.simple-consumer-reconnectLimit", 3);
+ this.clientId = config.getProperty("group.id",
"flink-kafka-consumer-legacy-" + broker.id());
Review comment:
Hi @azagrebin,
group.id and client.id are two separate concepts, it has different meanings.
I mean that there's no real meaning to the name of the client id as
"flink-kafka-consumer-legacy-" + broker.id(). As different topics consumed by
different consumers are all named with this value, it is impossible to
distinguish and difficult to troubleshoot.
I would suggest to open the client id to the user, and its default value set
"group.id value".
thanks.
Cheers
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
[GitHub] maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify clientId to groupId in flink-connector-kafka-0.8
maqingxiang commented on a change in pull request #5304: [FLINK-8290]Modify
clientId to groupId in flink-connector-kafka-0.8
URL: https://github.com/apache/flink/pull/5304#discussion_r208448741
##
File path:
flink-connectors/flink-connector-kafka-0.8/src/main/java/org/apache/flink/streaming/connectors/kafka/internals/SimpleConsumerThread.java
##
@@ -123,6 +124,7 @@ public SimpleConsumerThread(
this.fetchSize = getInt(config, "fetch.message.max.bytes",
1048576);
this.bufferSize = getInt(config, "socket.receive.buffer.bytes",
65536);
this.reconnectLimit = getInt(config,
"flink.simple-consumer-reconnectLimit", 3);
+ this.clientId = config.getProperty("group.id",
"flink-kafka-consumer-legacy-" + broker.id());
Review comment:
Hi, @azagrebin
group.id is a concept of kafka, used to describe consumers property, its
official description is "A string that uniquely identifies the group of
consumer processes to which this consumer belongs. By setting the same group id
multiple processes indicate that they are all part of the same consumer group".
client.id is a concept of SimpleConsumer, its official descriptionis "The
client id is a user-specified string sent in each request to help trace calls.
It should logically identify the application making the request". its default
value is "group.id value"
thanks.
This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
With regards,
Apache Git Services
