You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "FrankYang0529 (via GitHub)" <gi...@apache.org> on 2025/06/04 15:09:17 UTC
[PR] KAFKA-19369: Add group.share.assignors config and integration test (wip) [kafka]
FrankYang0529 opened a new pull request, #19900:
URL: https://github.com/apache/kafka/pull/19900
* Add `group.share.assignors` config to `GroupCoordinatorConfig`.
* Send `rackId` in share group heartbeat request if it's not null.
* Add integration test `testShareConsumerWithRackAwareAssignor`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "DL1231 (via GitHub)" <gi...@apache.org>.
DL1231 commented on code in PR #19900:
URL: https://github.com/apache/kafka/pull/19900#discussion_r2131283934
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -550,6 +565,43 @@ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
return assignors;
}
+ protected List<ShareGroupPartitionAssignor> shareGroupAssignors(
+ AbstractConfig config
+ ) {
+ List<ShareGroupPartitionAssignor> assignors = new ArrayList<>();
+
+ try {
+ for (String kclass : config.getList(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG)) {
+ ShareGroupPartitionAssignor assignor;
+
+ if (Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
+ assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
+ } else {
+ try {
+ assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
+ } catch (ClassNotFoundException e) {
+ throw new KafkaException("Class " + kclass + " cannot be found", e);
+ } catch (ClassCastException e) {
+ throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
+ }
+ }
Review Comment:
```suggestion
ShareGroupPartitionAssignor assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
if (!Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
try {
assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
} catch (ClassNotFoundException e) {
throw new KafkaException("Class " + kclass + " cannot be found", e);
} catch (ClassCastException e) {
throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
}
}
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19900:
URL: https://github.com/apache/kafka/pull/19900#discussion_r2131486262
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -550,6 +565,43 @@ protected List<ConsumerGroupPartitionAssignor> consumerGroupAssignors(
return assignors;
}
+ protected List<ShareGroupPartitionAssignor> shareGroupAssignors(
+ AbstractConfig config
+ ) {
+ List<ShareGroupPartitionAssignor> assignors = new ArrayList<>();
+
+ try {
+ for (String kclass : config.getList(GroupCoordinatorConfig.SHARE_GROUP_ASSIGNORS_CONFIG)) {
+ ShareGroupPartitionAssignor assignor;
+
+ if (Objects.equals(kclass, SHARE_GROUP_ASSIGNORS_DEFAULT)) {
+ assignor = SHARE_GROUP_BUILTIN_ASSIGNOR;
+ } else {
+ try {
+ assignor = Utils.newInstance(kclass, ShareGroupPartitionAssignor.class);
+ } catch (ClassNotFoundException e) {
+ throw new KafkaException("Class " + kclass + " cannot be found", e);
+ } catch (ClassCastException e) {
+ throw new KafkaException(kclass + " is not an instance of " + ShareGroupPartitionAssignor.class.getName());
+ }
+ }
Review Comment:
Thanks for the review. Updated it.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on code in PR #19900:
URL: https://github.com/apache/kafka/pull/19900#discussion_r2131702416
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -235,6 +238,13 @@ public class GroupCoordinatorConfig {
public static final int SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DEFAULT = 15000;
public static final String SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DOC = "The maximum heartbeat interval for share group members.";
+ private static final ShareGroupPartitionAssignor SHARE_GROUP_BUILTIN_ASSIGNOR = new SimpleAssignor();
+ public static final String SHARE_GROUP_ASSIGNORS_CONFIG = "group.share.assignors";
+ public static final String SHARE_GROUP_ASSIGNORS_DOC = "The server-side assignors as a list of full class names. " +
Review Comment:
Agree. I will update the PR later. Thanks.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test (wip) [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #19900:
URL: https://github.com/apache/kafka/pull/19900#issuecomment-2940407323
I will remove wip and rebase trunk after https://github.com/apache/kafka/pull/19856 is merged.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield merged PR #19900:
URL: https://github.com/apache/kafka/pull/19900
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "FrankYang0529 (via GitHub)" <gi...@apache.org>.
FrankYang0529 commented on PR #19900:
URL: https://github.com/apache/kafka/pull/19900#issuecomment-2943027971
The flaky test `PlaintextConsumerTest#testCloseLeavesGroupOnInterrupt` is not related to this PR.
https://issues.apache.org/jira/browse/KAFKA-18031
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19369: Add group.share.assignors config and integration test [kafka]
Posted by "AndrewJSchofield (via GitHub)" <gi...@apache.org>.
AndrewJSchofield commented on code in PR #19900:
URL: https://github.com/apache/kafka/pull/19900#discussion_r2131694269
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorConfig.java:
##########
@@ -235,6 +238,13 @@ public class GroupCoordinatorConfig {
public static final int SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DEFAULT = 15000;
public static final String SHARE_GROUP_MAX_HEARTBEAT_INTERVAL_MS_DOC = "The maximum heartbeat interval for share group members.";
+ private static final ShareGroupPartitionAssignor SHARE_GROUP_BUILTIN_ASSIGNOR = new SimpleAssignor();
+ public static final String SHARE_GROUP_ASSIGNORS_CONFIG = "group.share.assignors";
+ public static final String SHARE_GROUP_ASSIGNORS_DOC = "The server-side assignors as a list of full class names. " +
Review Comment:
I think you've done a slightly different thing here than it says in the KIP. I think it's an improvement so I want to adjust the KIP too. The doc should say `"The server-side assignors as a list of either names for built-in assignors or full class names for custom assignors. The list must contain only a single entry which is used by all groups. The supported built-in assignors are: " + SHARE_GROUP_BUILTIN_ASSIGNOR.name() + "."`. I don't think there's any need to include the sentence about future plans. That's more helpful as commentary in the KIP. In this way, I think we have a futureproof behaviour and accurate documentation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org