You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "TaiJuWu (via GitHub)" <gi...@apache.org> on 2024/09/04 04:11:04 UTC
[PR] [Draft] KAFKA-17456 [kafka]
TaiJuWu opened a new pull request, #17080:
URL: https://github.com/apache/kafka/pull/17080
*More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.*
*Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.*
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #17080:
URL: https://github.com/apache/kafka/pull/17080#issuecomment-2328442012
> @TaiJuWu Please fix the PR title, correct JIRA number is `17465`
Thanks for your remainder, I will address your comments later.
--
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-17456: Make getMembersFromGroup be non-blocking [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on PR #17080:
URL: https://github.com/apache/kafka/pull/17080#issuecomment-2328417299
@TaiJuWu
Please fix the PR title, correct JIRA number is `17465`
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 merged PR #17080:
URL: https://github.com/apache/kafka/pull/17080
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1760478308
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4144,26 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+ if (!options.removeAll()) {
Review Comment:
Done.
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1744717047
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,27 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex));
+ }
Review Comment:
My question is:
In the original logic, if an exception is caught, the method stops executing and throws the exception. However, in your PR, the handling process seems to differ from this.
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1744727700
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,27 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex));
+ }
Review Comment:
Thanks, you are right😺
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1769485396
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4140,27 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+
+ KafkaFutureImpl<List<MemberIdentity>> f;
Review Comment:
All comments are addressed, 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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1760478308
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4144,26 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+ if (!options.removeAll()) {
Review Comment:
Done. Thanks for review!
--
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-17456: Make getMembersFromGroup be non-blocking [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1743420113
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,30 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(ex);
+ }
+ Collection<MemberDescription> members = res.members();
+ List<MemberIdentity> membersToRemove = new ArrayList<>();
+ for (final MemberDescription member : members) {
+ MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
- if (member.groupInstanceId().isPresent()) {
- memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
- } else {
- memberIdentity.setMemberId(member.consumerId());
+ if (member.groupInstanceId().isPresent()) {
+ memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
+ } else {
+ memberIdentity.setMemberId(member.consumerId());
+ }
Review Comment:
nit: How about leverage api of `Optional` ?
```java
member.groupInstanceId()
.map(memberIdentity::setGroupInstanceId)
.orElseGet(() -> memberIdentity.setMemberId(member.consumerId()));
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,30 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(ex);
+ }
+ Collection<MemberDescription> members = res.members();
+ List<MemberIdentity> membersToRemove = new ArrayList<>();
+ for (final MemberDescription member : members) {
Review Comment:
`for (final MemberDescription member : res.members())`
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,30 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(ex);
+ }
Review Comment:
Do we need to keep executing following logic even if we encounter exception ?
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1763983367
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,22 +4144,33 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+ if (!options.removeAll()) {
+ List<MemberIdentity> members = options.members().stream()
+ .map(m -> m.toMemberIdentity().setReason(reason))
+ .collect(Collectors.toList());
+ handleRemoveMembersFromConsumerGroupAndInvokeDriver(groupId, members, options.timeoutMs, future);
+ return new RemoveMembersFromConsumerGroupResult(future.get(CoordinatorKey.byGroupId(groupId)), options.members());
}
- SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
- RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
- RemoveMembersFromConsumerGroupHandler handler = new RemoveMembersFromConsumerGroupHandler(groupId, members, logContext);
- invokeDriver(handler, future, options.timeoutMs);
+ getMembersFromGroup(groupId, reason).whenComplete((members, ex) -> {
Review Comment:
how about:
```java
KafkaFutureImpl<List<MemberIdentity>> f;
if (options.removeAll()) f = getMembersFromGroup(groupId, reason);
else {
f = new KafkaFutureImpl<>();
f.complete(options.members().stream()
.map(m -> m.toMemberIdentity().setReason(reason))
.collect(Collectors.toList()));
}
f.whenComplete((members, ex) -> {
if (ex != null) {
future.completeExceptionally(Collections.singletonMap(CoordinatorKey.byGroupId(groupId), ex));
} else {
RemoveMembersFromConsumerGroupHandler handler = new RemoveMembersFromConsumerGroupHandler(groupId, members, logContext);
invokeDriver(handler, future, options.timeoutMs());
}
});
```
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1759637125
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4144,26 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+ if (!options.removeAll()) {
Review Comment:
Could you please try to reduce duplicate code?
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1769326791
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,28 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
-
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- if (member.groupInstanceId().isPresent()) {
- memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex));
} else {
- memberIdentity.setMemberId(member.consumerId());
+ List<MemberIdentity> membersToRemove = new ArrayList<>();
+ for (final MemberDescription member : res.members()) {
+ MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
Review Comment:
If you prefer to use lambda function, please try to eliminate the local variable. for example:`
```java
membersToRemove.add(member.groupInstanceId()
.map(id -> new MemberIdentity().setGroupInstanceId(id))
.orElseGet(() -> new MemberIdentity().setMemberId(member.consumerId()))
.setReason(reason));
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,28 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
-
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- if (member.groupInstanceId().isPresent()) {
- memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex));
} else {
- memberIdentity.setMemberId(member.consumerId());
+ List<MemberIdentity> membersToRemove = new ArrayList<>();
Review Comment:
Could you please initialize the size by `res.members()`?
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1764204194
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,22 +4144,33 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+ if (!options.removeAll()) {
+ List<MemberIdentity> members = options.members().stream()
+ .map(m -> m.toMemberIdentity().setReason(reason))
+ .collect(Collectors.toList());
+ handleRemoveMembersFromConsumerGroupAndInvokeDriver(groupId, members, options.timeoutMs, future);
+ return new RemoveMembersFromConsumerGroupResult(future.get(CoordinatorKey.byGroupId(groupId)), options.members());
}
- SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
- RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
- RemoveMembersFromConsumerGroupHandler handler = new RemoveMembersFromConsumerGroupHandler(groupId, members, logContext);
- invokeDriver(handler, future, options.timeoutMs);
+ getMembersFromGroup(groupId, reason).whenComplete((members, ex) -> {
Review Comment:
Done. Thanks for review.
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1769451569
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,28 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
-
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- if (member.groupInstanceId().isPresent()) {
- memberIdentity.setGroupInstanceId(member.groupInstanceId().get());
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex));
} else {
- memberIdentity.setMemberId(member.consumerId());
+ List<MemberIdentity> membersToRemove = new ArrayList<>();
+ for (final MemberDescription member : res.members()) {
+ MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
Review Comment:
Hi @chia7712 , thanks for review.
I rewrite this part, please take a look.
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1769459881
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4140,27 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+
+ KafkaFutureImpl<List<MemberIdentity>> f;
+ if (options.removeAll()) f = getMembersFromGroup(groupId, reason);
Review Comment:
nit: I would recommend not omitting the curly braces, or at the very least, it should break the line.
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4143,19 +4140,27 @@ public RemoveMembersFromConsumerGroupResult removeMembersFromConsumerGroup(Strin
String reason = options.reason() == null || options.reason().isEmpty() ?
DEFAULT_LEAVE_GROUP_REASON : JoinGroupRequest.maybeTruncateReason(options.reason());
- List<MemberIdentity> members;
- if (options.removeAll()) {
- members = getMembersFromGroup(groupId, reason);
- } else {
- members = options.members().stream()
- .map(m -> m.toMemberIdentity().setReason(reason))
- .collect(Collectors.toList());
+ final SimpleAdminApiFuture<CoordinatorKey, Map<MemberIdentity, Errors>> future =
+ RemoveMembersFromConsumerGroupHandler.newFuture(groupId);
+
+ KafkaFutureImpl<List<MemberIdentity>> f;
Review Comment:
Please rename this variable.
--
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-17456: Make getMembersFromGroup be non-blocking [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #17080:
URL: https://github.com/apache/kafka/pull/17080#discussion_r1743398593
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -4114,27 +4114,30 @@ private Integer nodeFor(ConfigResource resource) {
}
}
- private List<MemberIdentity> getMembersFromGroup(String groupId, String reason) {
- Collection<MemberDescription> members;
- try {
- members = describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).get().members();
- } catch (Exception ex) {
- throw new KafkaException("Encounter exception when trying to get members from group: " + groupId, ex);
- }
+ private KafkaFutureImpl<List<MemberIdentity>> getMembersFromGroup(String groupId, String reason) {
+ KafkaFutureImpl<List<MemberIdentity>> future = new KafkaFutureImpl<>();
- List<MemberIdentity> membersToRemove = new ArrayList<>();
- for (final MemberDescription member : members) {
- MemberIdentity memberIdentity = new MemberIdentity().setReason(reason);
+ describeConsumerGroups(Collections.singleton(groupId)).describedGroups().get(groupId).whenComplete((res, ex) -> {
+ if (ex != null) {
+ future.completeExceptionally(ex);
+ }
Review Comment:
Do we need to keep executing following logic even if we encounter exception ?
In original one, it seems that the exception will be wrapped as `KafkaException` and throw.
--
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-17465: Make getMembersFromGroup be non-blocking [kafka]
Posted by "TaiJuWu (via GitHub)" <gi...@apache.org>.
TaiJuWu commented on PR #17080:
URL: https://github.com/apache/kafka/pull/17080#issuecomment-2328888796
> Hi @TaiJuWu I have few comments, PTAL
Hi @frankvicky ,thanks for review, all comments are addressed.
--
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