dajac commented on code in PR #15152:
URL: https://github.com/apache/kafka/pull/15152#discussion_r1447420131
##########
core/src/main/scala/kafka/coordinator/group/GroupMetadata.scala:
##########
@@ -622,7 +623,7 @@ private[group] class GroupMetadata(val groupId: String,
initialState: GroupState
}
def overview: GroupOverview = {
- GroupOverview(groupId, protocolType.getOrElse(""), state.toString)
+ GroupOverview(groupId, protocolType.getOrElse(""), state.toString,
"classic")
Review Comment:
nit: ditto about using GroupType.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -453,19 +454,31 @@ public Group group(String groupId, long committedOffset)
throws GroupIdNotFoundE
/**
* Get the Group List.
*
- * @param statesFilter The states of the groups we want to list.
- * If empty all groups are returned with their state.
- * @param committedOffset A specified committed offset corresponding to
this shard
+ * @param statesFilter The states of the groups we want to list.
+ * If empty, all groups are returned with their
state.
+ * @param typesFilter The types of the groups we want to list.
+ * If empty, all groups are returned with their
type.
+ * @param committedOffset A specified committed offset corresponding to
this shard.
*
* @return A list containing the ListGroupsResponseData.ListedGroup
*/
+ public List<ListGroupsResponseData.ListedGroup> listGroups(
+ List<String> statesFilter,
+ List<String> typesFilter,
+ long committedOffset
+ ) {
+ Predicate<Group> combinedFilter = group -> {
+ boolean stateCheck = statesFilter.isEmpty() ||
statesFilter.contains(group.stateAsString(committedOffset));
+ boolean typeCheck = typesFilter.isEmpty() ||
typesFilter.contains(group.type().toString());
Review Comment:
How do we handle the case insensitive?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -453,19 +454,31 @@ public Group group(String groupId, long committedOffset)
throws GroupIdNotFoundE
/**
* Get the Group List.
*
- * @param statesFilter The states of the groups we want to list.
- * If empty all groups are returned with their state.
- * @param committedOffset A specified committed offset corresponding to
this shard
+ * @param statesFilter The states of the groups we want to list.
+ * If empty, all groups are returned with their
state.
+ * @param typesFilter The types of the groups we want to list.
+ * If empty, all groups are returned with their
type.
+ * @param committedOffset A specified committed offset corresponding to
this shard.
*
* @return A list containing the ListGroupsResponseData.ListedGroup
*/
+ public List<ListGroupsResponseData.ListedGroup> listGroups(
+ List<String> statesFilter,
+ List<String> typesFilter,
Review Comment:
nit: Should we use Set for those two?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -9583,24 +9584,24 @@ public void
testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
@Test
public void testListGroups() {
Review Comment:
I don't see any test cases with group types. Should we add some?
##########
core/src/main/scala/kafka/coordinator/group/GroupCoordinator.scala:
##########
@@ -1105,16 +1105,18 @@ private[group] class GroupCoordinator(
}
}
- def handleListGroups(states: Set[String]): (Errors, List[GroupOverview]) = {
+ def handleListGroups(states: Set[String], groupTypes: Set[String]): (Errors,
List[GroupOverview]) = {
if (!isActive.get) {
(Errors.COORDINATOR_NOT_AVAILABLE, List[GroupOverview]())
} else {
val errorCode = if (groupManager.isLoading)
Errors.COORDINATOR_LOAD_IN_PROGRESS else Errors.NONE
- // if states is empty, return all groups
- val groups = if (states.isEmpty)
- groupManager.currentGroups
- else
- groupManager.currentGroups.filter(g =>
states.contains(g.summary.state))
+ // Filter groups based on states and groupTypes. If either is empty, it
won't filter on that criterion.
+ // If groupType is mentioned then no group is returned since the notion
of groupTypes doesn't exist in the
+ // old group coordinator.
+ val groups = groupManager.currentGroups.filter { g =>
+ (states.isEmpty || states.contains(g.summary.state)) &&
+ (groupTypes.isEmpty || groupTypes.contains("classic"))
Review Comment:
nit: Could we use `GroupType.CLASSIC` instead of hardcoding `classic` here?
##########
clients/src/main/java/org/apache/kafka/common/ConsumerGroupType.java:
##########
@@ -0,0 +1,50 @@
+/*
Review Comment:
It seems that we only use in tests. Could we use
`org.apache.kafka.coordinator.group.Group.GroupType` enum instead?
--
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]
