You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "dongnuo123 (via GitHub)" <gi...@apache.org> on 2023/09/27 14:35:35 UTC
[GitHub] [kafka] dongnuo123 opened a new pull request, #14462: KAFKA-14504: Implement DescribeGroups API
dongnuo123 opened a new pull request, #14462:
URL: https://github.com/apache/kafka/pull/14462
This PR implements DescribeGroups API according to the implementation in the old GroupCoordinator with the new model.
### JIRA
https://issues.apache.org/jira/browse/KAFKA-14504
### 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-14504: Implement DescribeGroups API [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1346196993
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
Review Comment:
add `testDescribeGroupsUncommittedGroup`
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
Review Comment:
added `testDescribeGroupsUncommittedGroup`
--
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-14504: Implement DescribeGroups API [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
dongnuo123 commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1346196163
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -521,6 +575,31 @@ GenericGroup getOrMaybeCreateGenericGroup(
}
}
+ /**
+ * Gets a generic group by committed offset.
+ *
+ * @param groupId The group id.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A GenericGroup.
+ * @throws GroupIdNotFoundException if the group does not exist or is not a generic group.
+ */
+ public GenericGroup getGenericGroupByCommittedOffset(
Review Comment:
done
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -751,6 +752,111 @@ public void testListGroupsFailedImmediately()
assertEquals(Collections.emptyList(), listGroupsResponseData.groups());
}
+ @Test
+ public void testDescribeGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 2;
+ service.startup(() -> partitionCount);
+
+ DescribeGroupsResponseData.DescribedGroup describedGroup1 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-1");
+ DescribeGroupsResponseData.DescribedGroup describedGroup2 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-2");
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
+ describedGroup1,
+ describedGroup2
+ );
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(describedGroup1)));
+
+ CompletableFuture<Object> describedGroupFuture = new CompletableFuture<>();
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 1)),
+ ArgumentMatchers.any()
+ )).thenReturn(describedGroupFuture);
+
+ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> future =
+ service.describeGroups(requestContext(ApiKeys.DESCRIBE_GROUPS), Arrays.asList("group-id-1", "group-id-2"));
+
+ assertFalse(future.isDone());
+ describedGroupFuture.complete(Collections.singletonList(describedGroup2));
+
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-14504: Implement DescribeGroups API [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14462:
URL: https://github.com/apache/kafka/pull/14462
--
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
[GitHub] [kafka] dajac commented on a diff in pull request #14462: KAFKA-14504: Implement DescribeGroups API
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1340758791
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -505,9 +508,52 @@ public CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> descri
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ final List<CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>>> futures =
+ new ArrayList<>(groupIds.size());
+ final Map<TopicPartition, List<String>> groupsByTopicPartition = new HashMap<>();
+ groupIds.forEach(groupId -> {
+ // For backwards compatibility, we support DescribeGroups for the empty group id.
+ if (groupId == null) {
+ futures.add(CompletableFuture.completedFuture(Collections.singletonList(
+ new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId(null)
+ .setErrorCode(Errors.INVALID_GROUP_ID.code())
+ )));
+ } else {
+ final TopicPartition topicPartition = topicPartitionFor(groupId);
+ groupsByTopicPartition
+ .computeIfAbsent(topicPartition, __ -> new ArrayList<>())
+ .add(groupId);
+ }
+ });
+
+ groupsByTopicPartition.forEach((topicPartition, groupList) -> {
+ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> future =
+ runtime.scheduleReadOperation(
+ "describe-groups",
+ topicPartition,
+ (coordinator, __) -> coordinator.describeGroups(context, groupList)
Review Comment:
@dongnuo123 We cannot ignore the second argument received here. Otherwise, we would expose uncommitted state. Take a look at how we did it for the list group api.
--
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-14504: Implement DescribeGroups API [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1345506634
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
+ assertEquals(responseData.protocolName(), describedGroups.get(0).protocolData());
+ assertEquals(responseData.members().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
+ }
+
+ @Test
+ public void testDescribeGroupStableForStaticMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ context.staticMembersJoinAndRebalance(
+ "group-id",
+ "leader-instance-id",
+ "follower-instance-id"
+ );
+ GenericGroup group = context.groupMetadataManager.getOrMaybeCreateGenericGroup("group-id", false);
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals("consumer", describedGroups.get(0).protocolType());
+ assertEquals("range", describedGroups.get(0).protocolData());
+ assertEquals(group.allMembers().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
+ assertEquals(group.allMembers().stream().map(member -> member.groupInstanceId().get()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.groupInstanceId()).collect(Collectors.toList())
+ );
+ }
+
+ @Test
+ public void testDescribeGroupRebalancing() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ GenericGroup group = context.createGenericGroup("group-id");
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteJoin(
+ new JoinGroupRequestBuilder()
+ .withGroupId("group-id")
+ .withMemberId(UNKNOWN_MEMBER_ID)
+ .withDefaultProtocolTypeAndProtocols()
+ .build()
+ );
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertTrue(group.isInState(COMPLETING_REBALANCE));
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
+ assertEquals("", describedGroups.get(0).protocolData());
+ assertTrue(describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ .contains(responseData.memberId()));
+ describedGroups.get(0).members().forEach(member -> assertTrue(member.memberMetadata().length == 0));
+ describedGroups.get(0).members().forEach(member -> assertTrue(member.memberAssignment().length == 0));
+ }
+
+ @Test
+ public void testDescribeGroupsGroupIdNotFoundException() {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ context.createSnapshot();
Review Comment:
nit: Instead of doing this, would it make sense to create the first snapshot in the constructor of the test context?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
+ assertEquals(responseData.protocolName(), describedGroups.get(0).protocolData());
+ assertEquals(responseData.members().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
Review Comment:
nit:
```
assertEquals(
responseData.members().stream().map(member -> member.memberId()).collect(Collectors.toList()),
describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
);
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
Review Comment:
It would be great if we could add a test similar to `testListGroups` which verifies that the committed offset works as expected.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
+ assertEquals(responseData.protocolName(), describedGroups.get(0).protocolData());
+ assertEquals(responseData.members().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
+ }
+
+ @Test
+ public void testDescribeGroupStableForStaticMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ context.staticMembersJoinAndRebalance(
+ "group-id",
+ "leader-instance-id",
+ "follower-instance-id"
+ );
+ GenericGroup group = context.groupMetadataManager.getOrMaybeCreateGenericGroup("group-id", false);
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals("consumer", describedGroups.get(0).protocolType());
+ assertEquals("range", describedGroups.get(0).protocolData());
+ assertEquals(group.allMembers().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
+ assertEquals(group.allMembers().stream().map(member -> member.groupInstanceId().get()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.groupInstanceId()).collect(Collectors.toList())
+ );
Review Comment:
nit: format as before.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
+ assertEquals(responseData.protocolName(), describedGroups.get(0).protocolData());
+ assertEquals(responseData.members().stream().map(member -> member.memberId()).collect(Collectors.toList()),
+ describedGroups.get(0).members().stream().map(member -> member.memberId()).collect(Collectors.toList())
+ );
+ }
+
+ @Test
+ public void testDescribeGroupStableForStaticMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ context.staticMembersJoinAndRebalance(
+ "group-id",
+ "leader-instance-id",
+ "follower-instance-id"
+ );
+ GenericGroup group = context.groupMetadataManager.getOrMaybeCreateGenericGroup("group-id", false);
Review Comment:
nit: Let's move this one to right before the assertions as it is only used there.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -751,6 +752,111 @@ public void testListGroupsFailedImmediately()
assertEquals(Collections.emptyList(), listGroupsResponseData.groups());
}
+ @Test
+ public void testDescribeGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 2;
+ service.startup(() -> partitionCount);
+
+ DescribeGroupsResponseData.DescribedGroup describedGroup1 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-1");
+ DescribeGroupsResponseData.DescribedGroup describedGroup2 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-2");
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
+ describedGroup1,
+ describedGroup2
+ );
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(describedGroup1)));
+
+ CompletableFuture<Object> describedGroupFuture = new CompletableFuture<>();
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 1)),
+ ArgumentMatchers.any()
+ )).thenReturn(describedGroupFuture);
+
+ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> future =
+ service.describeGroups(requestContext(ApiKeys.DESCRIBE_GROUPS), Arrays.asList("group-id-1", "group-id-2"));
+
+ assertFalse(future.isDone());
+ describedGroupFuture.complete(Collections.singletonList(describedGroup2));
+
Review Comment:
nit: Should we remove this empty line?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -521,6 +575,31 @@ GenericGroup getOrMaybeCreateGenericGroup(
}
}
+ /**
+ * Gets a generic group by committed offset.
+ *
+ * @param groupId The group id.
+ * @param committedOffset A specified committed offset corresponding to this shard.
+ *
+ * @return A GenericGroup.
+ * @throws GroupIdNotFoundException if the group does not exist or is not a generic group.
+ */
+ public GenericGroup getGenericGroupByCommittedOffset(
Review Comment:
nit: We could call it `genericGroup` to be aligned with `group`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8686,6 +8695,88 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(responseData.protocolType(), describedGroups.get(0).protocolType());
Review Comment:
Would it be possible to validate all the fields? One way would be to construct the expected response and to use assertEquals? This comment applies to all the tests.
--
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-14504: Implement DescribeGroups API [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1347070423
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -1210,6 +1229,11 @@ private void replay(
lastWrittenOffset++;
snapshotRegistry.getOrCreateSnapshot(lastWrittenOffset);
}
+
+ private void createSnapshot() {
+ lastWrittenOffset++;
+ snapshotRegistry.getOrCreateSnapshot(lastWrittenOffset);
+ }
Review Comment:
This is weird because we should only increment lastWrittenOffset when records are written. I think that we should remove it. See my other comments to avoid it.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8641,6 +8665,130 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
Review Comment:
The weirdness comes from the fact that joinGenericGroupAsDynamicMemberAndCompleteRebalance does not replay the generated records so the lastWrittenOffset is not updated and a new snapshot is not created. I tried to add the logic to replay them but that breaks a few tests because they assume that the GenericGroup object is updated but it is replaced when records are replayed. We should fix this but this is for another PR. Instead of doing this, we could just replay the relevant records to build the state that we want. Example:
```
# Create first group.
context.replay(newGroupMetadataRecord(....));
# Commit it.
context.commit()
# Create second group.
context.replay(newGroupMetadataRecord(....));
# Verify that first group is returned and second one is returned as dead
# Commit
# Verify that both groups are returned
```
What do you think?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -1134,6 +1140,19 @@ public CoordinatorResult<LeaveGroupResponseData, Record> sendGenericGroupLeave(
return groupMetadataManager.genericGroupLeave(context, request);
}
+ private void testDescribeGroupsUncommittedGroup(String groupId) {
Review Comment:
nit: `verifyDescribeGroupsReturnsDeadGroup(...)`? We use verify for similar methods in the context and this method is not aware of the "uncommitted" so it should not be in the name in my opinion.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8641,6 +8665,130 @@ public void testListGroups() {
assertEquals(expectAllGroupMap, actualAllGroupMap);
}
+ @Test
+ public void testDescribeGroupStableForDynamicMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ JoinGroupResponseData responseData = context.joinGenericGroupAsDynamicMemberAndCompleteRebalance("group-id");
+ context.testDescribeGroupsUncommittedGroup("group-id");
+
+ context.createSnapshot();
+ context.commit();
+
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Collections.singletonList(
+ new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id")
+ .setGroupState(STABLE.toString())
+ .setProtocolType(responseData.protocolType())
+ .setProtocolData(responseData.protocolName())
+ .setMembers(responseData.members().stream().map(member ->
+ new DescribeGroupsResponseData.DescribedGroupMember()
+ .setMemberId(member.memberId())
+ .setGroupInstanceId(member.groupInstanceId())
+ .setClientId("client")
+ .setClientHost("localhost/127.0.0.1")
+ .setMemberMetadata(member.metadata())
+ ).collect(Collectors.toList()))
+ );
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(expectedDescribedGroups, describedGroups);
+ }
+
+ @Test
+ public void testDescribeGroupStableForStaticMember() throws Exception {
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .build();
+ context.staticMembersJoinAndRebalance(
+ "group-id",
+ "leader-instance-id",
+ "follower-instance-id"
+ );
+ context.testDescribeGroupsUncommittedGroup("group-id");
+
+ context.createSnapshot();
+ context.commit();
+
+ GenericGroup group = context.groupMetadataManager.getOrMaybeCreateGenericGroup("group-id", false);
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Collections.singletonList(
+ new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id")
+ .setGroupState(STABLE.toString())
+ .setProtocolType(group.protocolType().get())
+ .setProtocolData(group.protocolName().get())
+ .setMembers(group.allMembers().stream().map(member ->
+ new DescribeGroupsResponseData.DescribedGroupMember()
+ .setMemberId(member.memberId())
+ .setGroupInstanceId(member.groupInstanceId().get())
+ .setClientId(member.clientId())
+ .setClientHost(member.clientHost())
+ .setMemberMetadata(member.metadata(group.protocolName().get()))
+ .setMemberAssignment(member.assignment())
+ ).collect(Collectors.toList()))
+ );
+
+ List<DescribeGroupsResponseData.DescribedGroup> describedGroups =
+ context.describeGroups(Collections.singletonList("group-id"));
+
+ assertEquals(expectedDescribedGroups, describedGroups);
+ }
+
+ @Test
+ public void testDescribeGroupRebalancing() throws Exception {
Review Comment:
In this case, you can probably replay and then trigger a rebalance to get the state that you want.
--
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-14504: Implement DescribeGroups API [kafka]
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14462:
URL: https://github.com/apache/kafka/pull/14462#discussion_r1342814124
##########
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:
When the group does not exist, in the current code, we return `GroupCoordinator.DeadGroup` instead of returning a `GroupIdNotFoundException` exception. Do you confirm?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -751,6 +752,112 @@ public void testListGroupsFailedImmediately()
assertEquals(Collections.emptyList(), listGroupsResponseData.groups());
}
+ @Test
+ public void testDescribeGroups() throws Exception {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 2;
+ service.startup(() -> partitionCount);
+
+ DescribeGroupsResponseData.DescribedGroup describedGroup1 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-1");
+ DescribeGroupsResponseData.DescribedGroup describedGroup2 = new DescribeGroupsResponseData.DescribedGroup()
+ .setGroupId("group-id-2");
+ List<DescribeGroupsResponseData.DescribedGroup> expectedDescribedGroups = Arrays.asList(
+ describedGroup1,
+ describedGroup2
+ );
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(describedGroup1)));
+
+ CompletableFuture<Object> describedGroupFuture = new CompletableFuture<>();
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("describe-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 1)),
+ ArgumentMatchers.any()
+ )).thenReturn(describedGroupFuture);
+
+ CompletableFuture<List<DescribeGroupsResponseData.DescribedGroup>> future =
+ service.describeGroups(requestContext(ApiKeys.DESCRIBE_GROUPS), Arrays.asList("group-id-1", "group-id-2"));
+
+ assertFalse(future.isDone());
+ describedGroupFuture.complete(Collections.singletonList(describedGroup2));
+
+ assertTrue(future.get().containsAll(expectedDescribedGroups));
+ assertTrue(expectedDescribedGroups.containsAll(future.get()));
Review Comment:
nit: Could we use assertEquals here?
##########
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:
Could we define a method similar to `getOrMaybeCreateGenericGroup(groupId, false)`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroupMember.java:
##########
@@ -349,6 +350,25 @@ public boolean isNew() {
return isNew;
}
+ /**
+ * @return The described group member without metadata.
+ */
+ public DescribeGroupsResponseData.DescribedGroupMember describeNoMetadata() {
Review Comment:
Should we add simple unit tests for those two as well?
##########
clients/src/main/java/org/apache/kafka/common/requests/DescribeGroupsRequest.java:
##########
@@ -75,4 +77,16 @@ public AbstractResponse getErrorResponse(int throttleTimeMs, Throwable e) {
public static DescribeGroupsRequest parse(ByteBuffer buffer, short version) {
return new DescribeGroupsRequest(new DescribeGroupsRequestData(new ByteBufferAccessor(buffer), version), version);
}
+
+ public static List<DescribeGroupsResponseData.DescribedGroup> getErrorDescribedGroupList(
Review Comment:
Could we add a unit test for this one please?
--
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-14504: Implement DescribeGroups API [kafka]
Posted by "dongnuo123 (via GitHub)" <gi...@apache.org>.
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org