dongnuo123 commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1343272684
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -444,6 +445,66 @@ public List<ListGroupsResponseData.ListedGroup>
listGroups(List<String> statesFi
return groupStream.map(group ->
group.asListedGroup(committedOffset)).collect(Collectors.toList());
}
+ /**
+ * Handles a DescribeGroup request.
+ *
+ * @param groupIds The IDs of the groups to describe.
+ * @param committedOffset A specified committed offset corresponding to
this shard.
+ *
+ * @return A list containing the DescribeGroupsResponseData.DescribedGroup.
+ */
+ public List<DescribeGroupsResponseData.DescribedGroup> describeGroups(
+ List<String> groupIds,
+ long committedOffset
+ ) {
+ final List<DescribeGroupsResponseData.DescribedGroup> describedGroups
= new ArrayList<>();
+ groupIds.forEach(groupId -> {
+ try {
+ Group group = group(groupId, committedOffset);
+ if (group.type() != GENERIC) {
+ // We don't support upgrading/downgrading between
protocols at the moment, so
+ // we throw an exception if a group exists with the wrong
type.
+ throw new GroupIdNotFoundException(String.format("Group %s
is not a generic group.",
+ groupId));
+ }
+ GenericGroup genericGroup = (GenericGroup) group;
Review Comment:
Yeah, you're right. We should return a GroupCoordinator.DeadGroup
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
