You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "zhaohaidao (via GitHub)" <gi...@apache.org> on 2023/08/22 01:18:51 UTC
[GitHub] [kafka] zhaohaidao opened a new pull request, #14271: Kafka 14503: Implement ListGroups
zhaohaidao opened a new pull request, #14271:
URL: https://github.com/apache/kafka/pull/14271
Implement ListGroups
### 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
[GitHub] [kafka] dajac commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1307483188
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
Right. Then you must pass `lastCommittedOffset` to `listGroups` as well and use it to query the timeline data structures.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314218928
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -309,6 +311,21 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param context The request context.
+ * @param request The ListGroups request.
+ * @return A Result containing the ListGroupsResponseData response
+ */
+ public ListGroupsResponseData listGroups(
+ RequestContext context,
+ ListGroupsRequestData request,
+ long committedOffset
+ ) throws ApiException {
Review Comment:
I tried to simplify the function signature based on my own understanding. The reason why the function return value here is consistent with the current one is that when an exception is thrown, the final result with an error code can be directly returned. Otherwise, the Runtime exception can only be thrown upward in completeExceptionaly. (a list of ListedGroup cannot pass in an error code). Then when handling the exception, you need to parse the cause of the RuntimeException, which I'm not sure is appropriate. Do you have any suggestions?
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1708137239
> @zhaohaidao There is a bunch of conflicts with `trunk`. Could you please fix them?
@dajac No problem, I will submit the fixed version 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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1323254383
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +606,85 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+
+ );
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(expectedResults));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ assertEquals(new ListGroupsResponseData().setGroups(expectedResults), responseFuture.get(5, TimeUnit.SECONDS));
+ }
+
+ private void testListGroupsFailedWithException(Throwable t, ListGroupsResponseData expectResponseData)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(t));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ assertEquals(expectResponseData, responseFuture.get(5, TimeUnit.SECONDS));
+
+ }
+
+ @Test
+ public void testListGroupsFutureFailed() throws InterruptedException, ExecutionException, TimeoutException {
+ for (Errors errors : Errors.values()) {
Review Comment:
I updated this test case based on my own understanding. Please help to see if it meets expectations.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1716017290
Hey, @dajac Review comments have been addressed, please take a look when you have time.
--
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 pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1701313706
@zhaohaidao Those are mainly flaky tests. There is one comment which is not addressed. Could you look into 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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314217880
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
+ */
+ public ListGroupsResponseData listGroups(ListGroupsRequestData request, long committedOffset) {
+ Stream<Group> groupStream = groups.values(committedOffset).stream();
+ List<String> statesFilter = request.statesFilter();
+ if (!statesFilter.isEmpty()) {
+ groupStream = groupStream.filter(group -> statesFilter.contains(group.stateAsString()));
+ }
+ return new ListGroupsResponseData().setGroups(groupStream.map(Group::asListedGroup).collect(Collectors.toList()));
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
[GitHub] [kafka] dajac commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1717271418
@zhaohaidao Thanks for all the work you are doing on this. I really appreciate it. I have created a few jiras for the next steps:
* https://issues.apache.org/jira/browse/KAFKA-15460
* https://issues.apache.org/jira/browse/KAFKA-15461
* https://issues.apache.org/jira/browse/KAFKA-15462
I have assigned all of them to you for now as you were interested in pursuing this work. Feel free to unassign them if you don't want or plan to tackle them.
From a priority perspective, tacking https://issues.apache.org/jira/browse/KAFKA-14504 before the above mentioned jiras would be really helpful as it will allow us to start end-to-end testing when we have all the existing APIs in place.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324884926
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -431,14 +434,17 @@ public void rollback() {
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
}
+ public void getOrCreateSnapshot() {
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
Review Comment:
This will cause testListGroups to fail because groupMetadataManager.listGroups always uses lastCommittedOffset to query. Putting getOrCreateSnapshot(lastWrittenOffset) at the beginning of the replay function will lead to the following result: the largest epoch in the snapshot is always 1 less than lastCommittedOffset. In this way, listGroups cannot find the expected results.
Putting getOrCreateSnapshot(lastWrittenOffset) at the end of the function, especially after lastWrittenOffset+1, all cases can pass, which is consistent with the semantics of CoordinatorRuntime, that is, always update the memory first, then update lastWrittenOffset, and use the latest lastWrittenOffset to update snapshot.
WDYT?
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1718105046
> @zhaohaidao Thanks for the update. I made another pass on it and I left a bunch of small comments. We should be able to merge it when they are addressed. Could you also rebase the PR to fix the conflicts?
Hey @dajac Except for a few comments that may still need to be discussed and are still in unresolved status, the other comments have been addressed. Please take a look when you have time.
--
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 pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1689042864
> @dajac hi, could you pls help me see the check error log. I don't quite understand the root cause and what should be fixed myself. The check error log is as followed
>
> > coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:129:1: Class Data Abstraction Coupling is 26 (max allowed is 25) classes [CompletableFuture, ConsumerGroup, ConsumerGroupHeartbeatResponseData, ConsumerGroupHeartbeatResponseData.Assignment, ConsumerGroupHeartbeatResponseData.TopicPartitions, ConsumerGroupMember.Builder, CoordinatorResult, CurrentAssignmentBuilder, FencedMemberEpochException, GenericGroup, GenericGroupMember, GroupIdNotFoundException, GroupMaxSizeReachedException, HeartbeatResponseData, InvalidRequestException, JoinGroupRequestProtocol, JoinGroupRequestProtocolCollection, JoinGroupResponseData, ListGroupsResponseData, SyncGroupResponseData, TargetAssignmentBuilder, TimelineHashMap, TimelineHashSet, UnknownMemberIdException, UnknownServerException, UnsupportedAssignorException].
@zhaohaidao It looks like we have hit the maximum number of imports. We could consider adding an exception for this. Have you tried 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
[GitHub] [kafka] dajac commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1307484913
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
Basically [here](https://github.com/apache/kafka/pull/14271/files#diff-00f0f81cf13e66781777d94f7d2e68a581663385c37e98792507f2294c91bb09R418), you need to use `lastCommittedOffset` to ensure that you only read committed state. I don't know if we support the Stream API but the other APIs should accept an argument called `epoch`. You can use `lastCommittedOffset` as the `epoch.`
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314217697
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -747,4 +749,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
+ }
Review Comment:
done.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
+ */
+ public ListGroupsResponseData listGroups(ListGroupsRequestData request, long committedOffset) {
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
[GitHub] [kafka] dajac commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1319725841
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -48,6 +49,16 @@ public String toString() {
*/
String stateAsString();
+ /**
+ * @return The {{@link GroupType}}'s String representation with committedOffset.
Review Comment:
nit: `based on the committed offset.`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +431,32 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ CompletableFuture<ListGroupsResponseData> future = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> results = new ArrayList<>();
Review Comment:
nit: Those two could be final as well.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +431,32 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ CompletableFuture<ListGroupsResponseData> future = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> results = new ArrayList<>();
+ final AtomicInteger cnt = new AtomicInteger(runtime.partitions().size());
+
+ for (TopicPartition tp : runtime.partitions()) {
Review Comment:
It seems to me that calling `partitions()` twice is not safe here because the number of partitions could change in between the two calls. I think that we should store it in order to avoid this race condition.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -306,6 +308,19 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param statesFilter The states of the groups we want to list. If empty all groups are returned with their state.
Review Comment:
nit: Add committedOffset as well.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -179,6 +181,23 @@ public String stateAsString() {
return state.get().toString();
}
+ /**
+ * @return The current state as a String with given committedOffset.
+ */
+ public String stateAsString(long committedOffset) {
+ return state.get(committedOffset).toString();
+ }
+
+ /**
+ * @return the group formatted as a list group response.
Review Comment:
nit: ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +606,85 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+
+ );
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
Review Comment:
Could we add more partitions to ensure that the logic to handle them work as expected?
##########
server-common/src/main/java/org/apache/kafka/server/util/FutureUtils.java:
##########
@@ -103,4 +104,12 @@ public static <T> CompletableFuture<T> failedFuture(Throwable ex) {
future.completeExceptionally(ex);
return future;
}
+
+ public static <T> void drainFutures(
Review Comment:
I suppose that we could remove this now, isn't it?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +422,17 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter and committedOffset.
+ */
Review Comment:
nit: Could we update the javadoc to full format? It would be great to document the arguments, etc. You have an example right below (`getOrMaybeCreateConsumerGroup`).
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -179,6 +181,23 @@ public String stateAsString() {
return state.get().toString();
}
+ /**
+ * @return The current state as a String with given committedOffset.
Review Comment:
nit: `based on the committed offset.` here as well.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/Group.java:
##########
@@ -48,6 +49,16 @@ public String toString() {
*/
String stateAsString();
+ /**
+ * @return The {{@link GroupType}}'s String representation with committedOffset.
+ */
+ String stateAsString(long committedOffset);
+
+ /**
+ * @return the group formatted as a list group response.
Review Comment:
nit: Should we also add `based on the committed offset.` here?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +606,85 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+
+ );
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(expectedResults));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ assertEquals(new ListGroupsResponseData().setGroups(expectedResults), responseFuture.get(5, TimeUnit.SECONDS));
+ }
+
+ private void testListGroupsFailedWithException(Throwable t, ListGroupsResponseData expectResponseData)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(t));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ assertEquals(expectResponseData, responseFuture.get(5, TimeUnit.SECONDS));
+
Review Comment:
nit: Remove empty line.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java:
##########
@@ -1169,7 +1179,7 @@ public List<JoinGroupResponseMember> currentGenericGroupMembers() {
/**
* @return the group formatted as a list group response.
Review Comment:
nit: ditto.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +606,85 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+
Review Comment:
nit: We can remove this empty line.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8999,4 +9052,4 @@ public SyncResult(
this.appendFuture = coordinatorResult.appendFuture();
}
}
-}
+}
Review Comment:
nit: Could we add an empty line at the end?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/generic/GenericGroup.java:
##########
@@ -242,6 +242,16 @@ public String stateAsString() {
return this.state.toString();
}
+ /**
+ * The state of this group with committedOffset.
Review Comment:
nit: `based on the committed offset.`.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +606,85 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+
+ );
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(expectedResults));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ assertEquals(new ListGroupsResponseData().setGroups(expectedResults), responseFuture.get(5, TimeUnit.SECONDS));
+ }
+
+ private void testListGroupsFailedWithException(Throwable t, ListGroupsResponseData expectResponseData)
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ service.startup(() -> 1);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(new TopicPartition("__consumer_offsets", 0)));
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 0)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(t));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ assertEquals(expectResponseData, responseFuture.get(5, TimeUnit.SECONDS));
+
+ }
+
+ @Test
+ public void testListGroupsFutureFailed() throws InterruptedException, ExecutionException, TimeoutException {
+ for (Errors errors : Errors.values()) {
Review Comment:
Testing all errors does not seem necessary to me. I think that we should test the following cases:
1. NOT_COORDINATOR is handled.
2. Other errors fail the future immediately even if not all the future are resolved. It would be great if you can have add a unresolved future in this case.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8597,36 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String genericGroupId = "generic-group-id";
+ String consumerGroupId = "consumer-group-id";
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.updateLastWrittenOffset(context.lastWrittenOffset);
+ GenericGroup genericGroup = context.createGenericGroup(genericGroupId);
+ ConsumerGroup consumerGroup = context.createConsumerGroup(consumerGroupId);
+ context.updateLastWrittenOffset(context.lastWrittenOffset + 2);
Review Comment:
In other tests, we use `replay()` to build the state. For instance, see `testConsumerGroupStates`. I wonder if we could reuse the same pattern here in order to keep the tests homogenous. What do you think?
Should we also test the state filtering somehow? One way would be to add a new member to the consumer group at some point.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1318745164
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +432,51 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new ArrayList<>();
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list-groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(request.statesFilter(), lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ throw new RuntimeException(exception);
+ }
+ if (exception instanceof CoordinatorLoadInProgressException) {
+ throw new RuntimeException(exception);
+ } else if (exception instanceof NotCoordinatorException) {
+ log.warn("ListGroups request {} hit a NotCoordinatorException exception: {}",
+ request, exception.getMessage());
+ return new ListGroupsResponseData().setGroups(Collections.emptyList());
+ } else {
+ return new ListGroupsResponseData().setErrorCode(Errors.forException(exception).code());
+ }
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ AtomicInteger succeedFutureCount = new AtomicInteger();
+ FutureUtils.drainFutures(futures, (data, t) -> {
+ synchronized (runtime) {
+ if (t != null) {
+ responseFuture.completeExceptionally(new UnknownServerException(t.getMessage()));
+ } else {
+ if (data.errorCode() != Errors.NONE.code()) {
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(data);
+ }
+ } else {
+ listedGroups.addAll(data.groups());
+ if (succeedFutureCount.addAndGet(1) == runtime.partitions().size()) {
+ responseFuture.complete(new ListGroupsResponseData().setGroups(listedGroups));
+ }
+ }
+ }
+ }
+ });
+ return responseFuture;
Review Comment:
Thank you for your suggestions. The code is much simpler and more elegant.
--
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] zhaohaidao commented on pull request #14271: Kafka 14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1687779192
@dajac hi, could you pls help me see the check error log. I don't quite understand the root cause and what should be fixed myself.
The check error log is as followed
> coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:129:1: Class Data Abstraction Coupling is 26 (max allowed is 25) classes [CompletableFuture, ConsumerGroup, ConsumerGroupHeartbeatResponseData, ConsumerGroupHeartbeatResponseData.Assignment, ConsumerGroupHeartbeatResponseData.TopicPartitions, ConsumerGroupMember.Builder, CoordinatorResult, CurrentAssignmentBuilder, FencedMemberEpochException, GenericGroup, GenericGroupMember, GroupIdNotFoundException, GroupMaxSizeReachedException, HeartbeatResponseData, InvalidRequestException, JoinGroupRequestProtocol, JoinGroupRequestProtocolCollection, JoinGroupResponseData, ListGroupsResponseData, SyncGroupResponseData, TargetAssignmentBuilder, TimelineHashMap, TimelineHashSet, UnknownMemberIdException, UnknownServerException, UnsupportedAssignorException].
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1715941885
> @zhaohaidao I just realized that we don't have any integration tests for this API. It would be great to add a few like I did [here](https://github.com/apache/kafka/pull/14353) for other APIs. Would you be interested by doing this? It could be done in a separate PR for sure.
> @zhaohaidao I just realized that we don't have any integration tests for this API. It would be great to add a few like I did [here](https://github.com/apache/kafka/pull/14353) for other APIs. Would you be interested by doing this? It could be done in a separate PR for sure.
hi, @dajac I'm interested in implementing integration tests, it would be a pleasure.
--
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 pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1715178703
Hey @zhaohaidao. Would you have an idea when you will be able to address the remaining comments? I would like to merge this one as soon as possible. We need KAFKA-14503 and KAFKA-14504 for end-to-end testing.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1701222669
> @zhaohaidao How is it going on this one?
@dajac Hi, the ci check is still failed and I have no idea what the error is . Could you pls help me to see this if you have time. The ci check result screen shot is as followed.
--
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 pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1713782755
@zhaohaidao I just realized that we don't have any integration tests for this API. It would be great to add a few like I did [here](https://github.com/apache/kafka/pull/14353) for other APIs. Would you be interested by doing this? It could be done in a separate PR for sure.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1307484913
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
Basically [here](https://github.com/apache/kafka/pull/14271/files#diff-00f0f81cf13e66781777d94f7d2e68a581663385c37e98792507f2294c91bb09R418), you need to use `lastCommittedOffset` to ensure that you only read committed state.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1710321306
> @zhaohaidao There is a bunch of conflicts with `trunk`. Could you please fix them?
@dajac hi, I have resolved conflicts with trunk and address your comments. could you pls continue to review if you are free. 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
[GitHub] [kafka] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1704074948
> @zhaohaidao Thanks for the update! I just made another pass and I left a few more comments for consideration. Could you take a look at them and let me know what you think?
Thanks very much for your comments. Your suggestion has prompted me to think more about error handling, and have a better understanding of the core design ideas of KIP-848.
I made an improvement according to your suggestion, please help to see if there is any need for improvement when you are free
--
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 pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1707768637
@zhaohaidao There is a bunch of conflicts with `trunk`. Could you please fix them?
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324188593
##########
checkstyle/suppressions.xml:
##########
@@ -327,7 +327,7 @@
files="(GroupMetadataManager|ConsumerGroupTest|GroupMetadataManagerTest).java"/>
<suppress checks="ClassFanOutComplexity"
files="(GroupMetadataManager|GroupMetadataManagerTest).java"/>
- <suppress checks="ParameterNumber"
+ <suppress checks="ParameterNumber|ClassDataAbstractionCoupling"
Review Comment:
nit: I just noticed that we already have `ClassDataAbstractionCouplingCheck` below. Could we add the relevant class(es) there instead of adding `ClassDataAbstractionCoupling` here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -306,6 +308,20 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @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
Review Comment:
nit: Could you please align the description of the params like we did for the other methods?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +422,22 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * 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
Review Comment:
nit: Same here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
Review Comment:
It seems to me that those assertions are not necessary as `assertEquals(expectedResults, actualResults);` already verifies everything, no?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
Review Comment:
nit: ditto about the format here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new NotCoordinatorException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
Review Comment:
ditto about those assertions.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new NotCoordinatorException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedImmediately()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
Review Comment:
nit: ditto about the format.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8999,4 +9096,5 @@ public SyncResult(
this.appendFuture = coordinatorResult.appendFuture();
}
}
-}
+
+}
Review Comment:
Sorry. I was not clear. I meant an empty line after the `}` not before.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -776,4 +802,5 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
Review Comment:
nit: Let's remove this empty line. This is not necessary.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
Review Comment:
small nit: Could we format this block as follow in order to follow the style in this file?
```
when(runtime.partitions()).thenReturn(Sets.newSet(
new TopicPartition("__consumer_offsets", 0),
new TopicPartition("__consumer_offsets", 1),
new TopicPartition("__consumer_offsets", 2)
));
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroupTest.java:
##########
@@ -632,6 +632,20 @@ public void testValidateOffsetCommit() {
group.validateOffsetCommit("member-id", "", 0);
}
+ @Test
+ public void testAsListedGroup() {
+ SnapshotRegistry snapshotRegistry = new SnapshotRegistry(new LogContext());
+ ConsumerGroup group = new ConsumerGroup(snapshotRegistry, "group-foo");
+ snapshotRegistry.getOrCreateSnapshot(0);
+ assertEquals(ConsumerGroup.ConsumerGroupState.EMPTY.toString(), group.stateAsString(0));
+ group.updateMember(new ConsumerGroupMember.Builder("member1")
+ .setSubscribedTopicNames(Collections.singletonList("foo"))
+ .build());
+ assertEquals(ConsumerGroup.ConsumerGroupState.EMPTY.toString(), group.stateAsString(0));
+ snapshotRegistry.getOrCreateSnapshot(1);
Review Comment:
nit: Could we move the snapshot creation to right after updating the member?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
+ context.getOrCreateSnapshot();
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList()).stream()
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ expectAllGroupMap = Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.ASSIGNING.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ actualAllGroupMap = context.sendListGroups(Collections.singletonList("Empty")).stream()
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ expectAllGroupMap = Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
Review Comment:
nit: Let's remove this empty line.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new NotCoordinatorException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedImmediately()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(new CompletableFuture<>());
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new CoordinatorLoadInProgressException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ ListGroupsResponseData listGroupsResponseData = responseFuture.get(5, TimeUnit.SECONDS);
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Collections.emptyList();
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = listGroupsResponseData.groups();
+
+ assertEquals(Errors.COORDINATOR_LOAD_IN_PROGRESS.code(), listGroupsResponseData.errorCode());
+ assertEquals(expectedResults, actualResults);
Review Comment:
nit: Would it make sense to inline expectedResults and actualResults? `assertEquals(Collections.emptyList(), listGroupsResponseData.groups())`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
Review Comment:
It would be great if you could put the following code here but without committing it yet. This would ensure that the last committed offset works as expected.
```
context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
.setSubscribedTopicNames(Collections.singletonList(fooTopicName))
.build()));
context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
```
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -431,14 +434,17 @@ public void rollback() {
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
}
+ public void getOrCreateSnapshot() {
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
Review Comment:
My intent when I wrote the test context was to hide the snapshots from the tests. I see why you need this. Because, snapshots are not always created... My bad. I wonder if we could create them all the time for all cases by adding `snapshotRegistry.getOrCreateSnapshot(lastWrittenOffset);` at the beginning of the `replay` method. Would this work?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
Review Comment:
nit: Following the style in this file, we would format this as follow:
```
Stream.of(
new ListGroupsResponseData.ListedGroup()
.setGroupId(genericGroup.groupId())
.setProtocolType(genericGroupType)
.setGroupState(EMPTY.toString()),
new ListGroupsResponseData.ListedGroup()
.setGroupId(consumerGroupId)
.setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
.setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString())
).collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
```
Could you also update the other similar blocks?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -663,7 +830,6 @@ public void testFetchAllOffsets(
createConfig(),
runtime
);
-
Review Comment:
nit: Let's revert this change.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
Review Comment:
We keep the commit here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
+ context.getOrCreateSnapshot();
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList()).stream()
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ expectAllGroupMap = Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.ASSIGNING.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
Review Comment:
ditto about the assertion.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
+ context.getOrCreateSnapshot();
+ actualAllGroupMap = context.sendListGroups(Collections.emptyList()).stream()
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ expectAllGroupMap = Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.ASSIGNING.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ actualAllGroupMap = context.sendListGroups(Collections.singletonList("Empty")).stream()
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ expectAllGroupMap = Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
Review Comment:
ditto about the assertion.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
Review Comment:
nit: I think that we could just do `assertEquals(expectAllGroupMap, actualAllGroupMap)`.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324870944
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new NotCoordinatorException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedImmediately()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(new CompletableFuture<>());
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new CoordinatorLoadInProgressException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ ListGroupsResponseData listGroupsResponseData = responseFuture.get(5, TimeUnit.SECONDS);
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Collections.emptyList();
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = listGroupsResponseData.groups();
+
+ assertEquals(Errors.COORDINATOR_LOAD_IN_PROGRESS.code(), listGroupsResponseData.errorCode());
+ assertEquals(expectedResults, actualResults);
Review Comment:
Indeed, this is more intuitive
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324869129
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
Review Comment:
Can you be more detailed? I don’t quite understand this comment.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
Review Comment:
Can you be more detailed? I don’t quite understand this comment.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324869392
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304575969
##########
clients/src/main/resources/common/message/ListGroupsRequest.json:
##########
@@ -23,11 +23,15 @@
// Version 3 is the first flexible version.
//
// Version 4 adds the StatesFilter field (KIP-518).
- "validVersions": "0-4",
+ //
+ // Version 5 adds the TypesFilter field (KIP-848).
+ "validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "StatesFilter", "type": "[]string", "versions": "4+",
"about": "The states of the groups we want to list. If empty all groups are returned with their state."
- }
+ },
+ { "name": "TypesFilter", "type": "[]string", "versions": "5+",
+ "about": "The types of the groups we want to list. If empty all groups are returned" }
Review Comment:
The test will use the code generated by the new protocol. If the protocol change is to be placed in the second pr, the corresponding test code also needs to be removed in this pr.
I don't know if my understanding is correct
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314219391
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
Review Comment:
done.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
Review Comment:
done.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314218928
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -309,6 +311,21 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param context The request context.
+ * @param request The ListGroups request.
+ * @return A Result containing the ListGroupsResponseData response
+ */
+ public ListGroupsResponseData listGroups(
+ RequestContext context,
+ ListGroupsRequestData request,
+ long committedOffset
+ ) throws ApiException {
Review Comment:
I tried to simplify the function signature based on my own understanding. The reason why the function return value here is consistent with the current one is that when an exception is thrown, the final result with an error code can be directly returned. Otherwise, the Runtime exception can only be thrown upward in completeExceptionaly. (List<ListsGroup> cannot pass in an error code). Then when handling the exception, you need to parse the cause of the RuntimeException, which I'm not sure is appropriate. Do you have any suggestions?
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1323262962
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,163 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+
+ @Test
+ public void testListGroupsFailedWithNotCoordinatorException()
+ throws InterruptedException, ExecutionException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < 2; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", 2)),
+ ArgumentMatchers.any()
+ )).thenReturn(FutureUtils.failedFuture(new NotCoordinatorException("")));
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
+ }
+ @Test
Review Comment:
nit: empty line between test.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1318751801
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +422,23 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter and committedOffset.
+ */
+ public List<ListGroupsResponseData.ListedGroup> listGroups(List<String> statesFilter, long committedOffset) {
+ Stream<Group> groupStream = groups.values(committedOffset).stream();
+ if (!statesFilter.isEmpty()) {
+ groupStream = groupStream.filter(group -> {
+ if (group instanceof ConsumerGroup) {
+ return statesFilter.contains(((ConsumerGroup) group).stateAsString(committedOffset));
+ } else {
+ return statesFilter.contains(group.stateAsString());
+ }
Review Comment:
make sense.
--
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 merged pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac merged PR #14271:
URL: https://github.com/apache/kafka/pull/14271
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304504593
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
I don’t know if I understand it right, here we need to replace __ with lastCommittedOffset, right?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
I don’t know if I understand it right, here we need to replace __ with lastCommittedOffset, 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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304456965
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
Review Comment:
> Don't we need to also implement this for `GenericGroup`?
GenericGroup has implemented the asListedGroup method. In fact, ConsumerGroup has this method first. In order to implement ListGroups, I abstracted it into an interface and implemented it in GenericGroup.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304456965
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
Review Comment:
> Don't we need to also implement this for `GenericGroup`?
GenericGroup has implemented the asListedGroup method. In fact, GenericGroup has this method first. In order to implement ListGroups, I abstracted it into an interface and implemented it in ConsumerGroup.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324791489
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
Review Comment:
Map is used here because the order of the list returned by the function may be different from the expected one.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1307482348
##########
clients/src/main/resources/common/message/ListGroupsRequest.json:
##########
@@ -23,11 +23,15 @@
// Version 3 is the first flexible version.
//
// Version 4 adds the StatesFilter field (KIP-518).
- "validVersions": "0-4",
+ //
+ // Version 5 adds the TypesFilter field (KIP-848).
+ "validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "StatesFilter", "type": "[]string", "versions": "4+",
"about": "The states of the groups we want to list. If empty all groups are returned with their state."
- }
+ },
+ { "name": "TypesFilter", "type": "[]string", "versions": "5+",
+ "about": "The types of the groups we want to list. If empty all groups are returned" }
Review Comment:
If the tests are related to the new filter, they should go to the next PR.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1691992754
> > @dajac hi, could you pls help me see the check error log. I don't quite understand the root cause and what should be fixed myself. The check error log is as followed
> > > coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:129:1: Class Data Abstraction Coupling is 26 (max allowed is 25) classes [CompletableFuture, ConsumerGroup, ConsumerGroupHeartbeatResponseData, ConsumerGroupHeartbeatResponseData.Assignment, ConsumerGroupHeartbeatResponseData.TopicPartitions, ConsumerGroupMember.Builder, CoordinatorResult, CurrentAssignmentBuilder, FencedMemberEpochException, GenericGroup, GenericGroupMember, GroupIdNotFoundException, GroupMaxSizeReachedException, HeartbeatResponseData, InvalidRequestException, JoinGroupRequestProtocol, JoinGroupRequestProtocolCollection, JoinGroupResponseData, ListGroupsResponseData, SyncGroupResponseData, TargetAssignmentBuilder, TimelineHashMap, TimelineHashSet, UnknownMemberIdException, UnknownServerException, UnsupportedAssignorException].
>
> @zhaohaidao It looks like we have hit the maximum number of imports. We could consider adding an exception for this. Have you tried this?
Thanks for your suggestion. I added an exception for this and it seems this check error has been fixed.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304529821
##########
clients/src/main/resources/common/message/ListGroupsRequest.json:
##########
@@ -23,11 +23,15 @@
// Version 3 is the first flexible version.
//
// Version 4 adds the StatesFilter field (KIP-518).
- "validVersions": "0-4",
+ //
+ // Version 5 adds the TypesFilter field (KIP-848).
+ "validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "StatesFilter", "type": "[]string", "versions": "4+",
"about": "The states of the groups we want to list. If empty all groups are returned with their state."
- }
+ },
+ { "name": "TypesFilter", "type": "[]string", "versions": "5+",
+ "about": "The types of the groups we want to list. If empty all groups are returned" }
Review Comment:
TODO: Add a second pr to update ListGroups protocol
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304488801
##########
clients/src/main/resources/common/message/ListGroupsRequest.json:
##########
@@ -23,11 +23,15 @@
// Version 3 is the first flexible version.
//
// Version 4 adds the StatesFilter field (KIP-518).
- "validVersions": "0-4",
+ //
+ // Version 5 adds the TypesFilter field (KIP-848).
+ "validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "StatesFilter", "type": "[]string", "versions": "4+",
"about": "The states of the groups we want to list. If empty all groups are returned with their state."
- }
+ },
+ { "name": "TypesFilter", "type": "[]string", "versions": "5+",
+ "about": "The types of the groups we want to list. If empty all groups are returned" }
Review Comment:
makes sense
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1715939170
> Hey @zhaohaidao. Would you have an idea when you will be able to address the remaining comments? I would like to merge this one as soon as possible. We need [KAFKA-14503](https://issues.apache.org/jira/browse/KAFKA-14503) and [KAFKA-14504](https://issues.apache.org/jira/browse/KAFKA-14504) for end-to-end testing.
Hey @dajac I'm working on the remaining parts and I'll launch a PR as soon as possible
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1323263745
##########
server-common/src/main/java/org/apache/kafka/server/util/FutureUtils.java:
##########
@@ -103,4 +103,5 @@ public static <T> CompletableFuture<T> failedFuture(Throwable ex) {
future.completeExceptionally(ex);
return future;
}
+
Review Comment:
nit: Remove this line.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1323252420
##########
server-common/src/main/java/org/apache/kafka/server/util/FutureUtils.java:
##########
@@ -103,4 +104,12 @@ public static <T> CompletableFuture<T> failedFuture(Throwable ex) {
future.completeExceptionally(ex);
return future;
}
+
+ public static <T> void drainFutures(
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
[GitHub] [kafka] dajac commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1700950618
@zhaohaidao How is it going on this one?
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314219340
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ futures.forEach(CompletableFuture::join);
+ for (CompletableFuture<ListGroupsResponseData> future : futures) {
+ try {
+ ListGroupsResponseData data = future.get();
+ if (data.errorCode() != Errors.NONE.code()) {
+ responseFuture.complete(data);
+ return responseFuture;
+ }
+ listedGroups.addAll(future.get().groups());
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, e.getMessage());
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(new ListGroupsResponseData()
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code()));
+ return responseFuture;
+ }
+ }
+ }
Review Comment:
I tried to simplify and improve a version based on my own understanding, but I’m not sure if it’s consistent with what you expected.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ futures.forEach(CompletableFuture::join);
+ for (CompletableFuture<ListGroupsResponseData> future : futures) {
+ try {
+ ListGroupsResponseData data = future.get();
+ if (data.errorCode() != Errors.NONE.code()) {
+ responseFuture.complete(data);
+ return responseFuture;
+ }
+ listedGroups.addAll(future.get().groups());
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, e.getMessage());
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(new ListGroupsResponseData()
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code()));
+ return responseFuture;
+ }
+ }
+ }
Review Comment:
I tried to simplify and improve a version based on my own understanding, but I’m not sure if it’s consistent with what you expected.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314944364
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -306,6 +308,19 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param statesFilter The states of the groups we want to list. If empty all groups are returned with their state.
+ * @return A Result containing the ListGroupsResponseData response
+ */
+ public ListGroupsResponseData listGroups(
+ List<String> statesFilter,
+ long committedOffset
+ ) throws ApiException {
+ return new ListGroupsResponseData().setGroups(groupMetadataManager.listGroups(statesFilter, committedOffset));
+ }
Review Comment:
I sill think that we should rather return the list of groups here and create `ListGroupsResponseData` one level up.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +432,51 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new ArrayList<>();
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list-groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(request.statesFilter(), lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ throw new RuntimeException(exception);
+ }
+ if (exception instanceof CoordinatorLoadInProgressException) {
+ throw new RuntimeException(exception);
+ } else if (exception instanceof NotCoordinatorException) {
+ log.warn("ListGroups request {} hit a NotCoordinatorException exception: {}",
+ request, exception.getMessage());
+ return new ListGroupsResponseData().setGroups(Collections.emptyList());
+ } else {
+ return new ListGroupsResponseData().setErrorCode(Errors.forException(exception).code());
+ }
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ AtomicInteger succeedFutureCount = new AtomicInteger();
+ FutureUtils.drainFutures(futures, (data, t) -> {
+ synchronized (runtime) {
+ if (t != null) {
+ responseFuture.completeExceptionally(new UnknownServerException(t.getMessage()));
+ } else {
+ if (data.errorCode() != Errors.NONE.code()) {
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(data);
+ }
+ } else {
+ listedGroups.addAll(data.groups());
+ if (succeedFutureCount.addAndGet(1) == runtime.partitions().size()) {
+ responseFuture.complete(new ListGroupsResponseData().setGroups(listedGroups));
+ }
+ }
+ }
+ }
+ });
+ return responseFuture;
Review Comment:
There are a few issues with this code.
1. Synchronising on `runtime` will create lock contention across all the callers of `listGroups`. We should rather use a local variable.
2. The error handling seems error prone to me. For instance, `NotCoordinatorException` exceptions are turned into `RuntimeException` exceptions and then turned into `UnknownServerException` if I understood it correctly. We lose the semantic along the way.
I think that we could take your idea further and combine the two main steps into one. I am thinking about something like this:
```
final List<TopicPartition> partitions = new ArrayList<>(runtime.partitions());
final CompletableFuture<ListGroupsResponseData> future = new CompletableFuture<>();
final List<ListGroupsResponseData.ListedGroup> results = new ArrayList<>();
final AtomicInteger cnt = new AtomicInteger(partitions.size());
for (TopicPartition partition : partitions) {
runtime.scheduleReadOperation(
"list-group",
partition,
(coordinator, lastCommittedOffset) -> coordinator.listGroups(request.statesFilter(), lastCommittedOffset)
).handle((groups, exception) -> {
if (exception == null) {
synchronized (results) {
results.addAll(groups);
}
} else {
if (!(exception instanceof NotCoordinatorException)) {
future.complete(new ListGroupsResponseData().setErrorCode(Errors.forException(exception).code()));
}
}
if (cnt.decrementAndGet() == 0 && !future.isDone()) {
future.complete(new ListGroupsResponseData().setGroups(results));
}
return null;
});
}
```
What do you think?
Note that I have not tested this code at all. I just wrote it here.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +422,23 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter and committedOffset.
+ */
+ public List<ListGroupsResponseData.ListedGroup> listGroups(List<String> statesFilter, long committedOffset) {
+ Stream<Group> groupStream = groups.values(committedOffset).stream();
+ if (!statesFilter.isEmpty()) {
+ groupStream = groupStream.filter(group -> {
+ if (group instanceof ConsumerGroup) {
+ return statesFilter.contains(((ConsumerGroup) group).stateAsString(committedOffset));
+ } else {
+ return statesFilter.contains(group.stateAsString());
+ }
Review Comment:
This is weird. I would rather push `stateAsString(committedOffset)` to the `Group` interface and use it.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ futures.forEach(CompletableFuture::join);
+ for (CompletableFuture<ListGroupsResponseData> future : futures) {
+ try {
+ ListGroupsResponseData data = future.get();
+ if (data.errorCode() != Errors.NONE.code()) {
+ responseFuture.complete(data);
+ return responseFuture;
+ }
+ listedGroups.addAll(future.get().groups());
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, e.getMessage());
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(new ListGroupsResponseData()
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code()));
+ return responseFuture;
+ }
+ }
+ }
Review Comment:
I was thinking about something like this:
```
public static <T> CompletableFuture<List<T>> allOf(List<CompletableFuture<T>> futures) {
return CompletableFuture
.allOf(futures.toArray(new CompletableFuture[0]))
.thenApply(__ -> {
final List<T> results = new ArrayList<>(futures.size());
for (CompletableFuture<T> future : futures) {
results.add(future.join());
}
return results;
});
}
```
I am actually not sure that it would help here, see my previous comment.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -179,6 +181,13 @@ public GroupType type() {
public String stateAsString() {
return state.get().toString();
}
+ /**
Review Comment:
nit: Let's put an empty line before this one.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -747,4 +756,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup(long committedOffset) {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.get(committedOffset).toString());
+ }
Review Comment:
nit: Could we move this method to before all the private method? For instance, we could put right after the new `stateAsString` method.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1719591231
> @zhaohaidao Thanks for the update and your patience. I just replied to your comments and left a few nits. The PR looks pretty good. We are almost there :).
Hey @dajac The latest comments have been addressed.
I’m not sure why, but there are a few comments I can’t reply to directly. In short, I have made modifications according to your suggestions. Please help me if you have time to see if there are still problems.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1325484973
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8603,6 +8614,86 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString())
+ ).collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap, actualAllGroupMap);
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
Review Comment:
I think that we can remove those two lines now that they are already executed earlier, no?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupCoordinatorServiceTest.java:
##########
@@ -599,6 +608,164 @@ public void testHeartbeatCoordinatorException() throws Exception {
);
}
+ @Test
+ public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
+ CoordinatorRuntime<GroupCoordinatorShard, Record> runtime = mockRuntime();
+ GroupCoordinatorService service = new GroupCoordinatorService(
+ new LogContext(),
+ createConfig(),
+ runtime
+ );
+ int partitionCount = 3;
+ service.startup(() -> partitionCount);
+
+ ListGroupsRequestData request = new ListGroupsRequestData();
+
+ List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group0")
+ .setGroupState("Stable")
+ .setProtocolType("protocol1"),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group1")
+ .setGroupState("Empty")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId("group2")
+ .setGroupState("Dead")
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ );
+ Map<String, List<ListGroupsResponseData.ListedGroup>> expectResultMap = new HashMap<>();
+ for (ListGroupsResponseData.ListedGroup result : expectedResults) {
+ expectResultMap.put(result.groupId(), Collections.singletonList(result));
+ }
+ when(runtime.partitions()).thenReturn(Sets.newSet(
+ new TopicPartition("__consumer_offsets", 0),
+ new TopicPartition("__consumer_offsets", 1),
+ new TopicPartition("__consumer_offsets", 2)));
+ for (int i = 0; i < partitionCount; i++) {
+ when(runtime.scheduleReadOperation(
+ ArgumentMatchers.eq("list-groups"),
+ ArgumentMatchers.eq(new TopicPartition("__consumer_offsets", i)),
+ ArgumentMatchers.any()
+ )).thenReturn(CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(i))));
+ }
+
+ CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
+ requestContext(ApiKeys.LIST_GROUPS),
+ request
+ );
+
+ List<ListGroupsResponseData.ListedGroup> actualResults = responseFuture.get(5, TimeUnit.SECONDS).groups();
+ assertEquals(expectedResults, actualResults);
+ assertEquals(expectResultMap.size(), actualResults.size());
+ for (ListGroupsResponseData.ListedGroup result : actualResults) {
+ assertEquals(expectResultMap.get(result.groupId()), Collections.singletonList(result));
+ }
Review Comment:
Hum... How could `assertEquals(expectedResults, actualResults)` work then? If the order of the `actualResults` changes, this assertion should fail from times to times, isn't it?
If the order is really an issue, how about using `assertEquals(new HashSet(expected), new HashSet(actual))`?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -455,7 +461,6 @@ public ConsumerGroupMember.MemberState consumerGroupMemberState(
public CoordinatorResult<ConsumerGroupHeartbeatResponseData, Record> consumerGroupHeartbeat(
ConsumerGroupHeartbeatRequestData request
) {
- snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
Review Comment:
nit: Let's remove this remaining empty line here.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -1139,6 +1148,7 @@ private ApiMessage messageOrNull(ApiMessageAndVersion apiMessageAndVersion) {
private void replay(
Record record
) {
+
Review Comment:
nit: Let's remove this empty line.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -434,14 +437,17 @@ public void rollback() {
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
}
+ public void getOrCreateSnapshot() {
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
+
public ConsumerGroup.ConsumerGroupState consumerGroupState(
String groupId
) {
return groupMetadataManager
.getOrMaybeCreateConsumerGroup(groupId, false)
.state();
}
-
Review Comment:
nit: Let's keep this empty line between the two methods.
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -431,14 +434,17 @@ public void rollback() {
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
}
+ public void getOrCreateSnapshot() {
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
Review Comment:
I think that your suggestion makes sense, thanks! I think that we could actually remove `getOrCreateSnapshot` now, right?
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8584,93 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String consumerGroupId = "consumer-group-id";
+ String genericGroupId = "generic-group-id";
+ String memberId1 = Uuid.randomUuid().toString();
+ String genericGroupType = "generic";
+ Uuid fooTopicId = Uuid.randomUuid();
+ String fooTopicName = "foo";
+
+ MockPartitionAssignor assignor = new MockPartitionAssignor("range");
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder()
+ .withAssignors(Collections.singletonList(assignor))
+ .withConsumerGroup(new ConsumerGroupBuilder(consumerGroupId, 10))
+ .build();
+ context.replay(newGroupMetadataRecord(
+ genericGroupId,
+ new GroupMetadataValue()
+ .setMembers(Collections.emptyList())
+ .setGeneration(2)
+ .setLeader(null)
+ .setProtocolType(genericGroupType)
+ .setProtocol("range")
+ .setCurrentStateTimestamp(context.time.milliseconds()),
+ MetadataVersion.latest()));
+ context.commit();
+ context.getOrCreateSnapshot();
+ GenericGroup genericGroup = context.groupMetadataManager.getOrMaybeCreateGenericGroup(genericGroupId, false);
+
+ Map<String, ListGroupsResponseData.ListedGroup> actualAllGroupMap =
+ context.sendListGroups(Collections.emptyList())
+ .stream().collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+ Map<String, ListGroupsResponseData.ListedGroup> expectAllGroupMap =
+ Stream.of(new ListGroupsResponseData.ListedGroup()
+ .setGroupId(genericGroup.groupId())
+ .setProtocolType(genericGroupType)
+ .setGroupState(EMPTY.toString()),
+ new ListGroupsResponseData.ListedGroup()
+ .setGroupId(consumerGroupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(ConsumerGroup.ConsumerGroupState.EMPTY.toString()))
+ .collect(Collectors.toMap(ListGroupsResponseData.ListedGroup::groupId, Function.identity()));
+
+ assertEquals(expectAllGroupMap.size(), actualAllGroupMap.size());
+ for (Map.Entry<String, ListGroupsResponseData.ListedGroup> entry : expectAllGroupMap.entrySet()) {
+ assertEquals(entry.getValue(), actualAllGroupMap.get(entry.getKey()));
+ }
+
+ context.replay(RecordHelpers.newMemberSubscriptionRecord(consumerGroupId, new ConsumerGroupMember.Builder(memberId1)
+ .setSubscribedTopicNames(Collections.singletonList(fooTopicName))
+ .build()));
+ context.replay(RecordHelpers.newGroupEpochRecord(consumerGroupId, 11));
+ context.commit();
Review Comment:
I just meant that we should keep the commit here when we move the replay of the records.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1312918434
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
Review Comment:
nit: Let's use `list-groups` to be consistent with the existing names.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
Review Comment:
nit: Let's remove `java.util` and `Collections.emptyList()` as they are not necessary.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
Review Comment:
I think that we need to think a little more about the different errors that we could get here. My understanding is that we fail the entire requests in two cases: 1) at least one partition is loading; and 2) there is an unexpected error (non KafkaException).
Here, we could get the following errors:
- NotCoordinatorException if `tp` is no longer active, failed, etc. In this case, we actually want to return an empty lit of groups.
- CoordinatorLoadingException if `tp` is being loaded. In this case, we want to fail the entire request.
- Unexpected Exception. In this case, we also want to fail the entire request.
Knowing this, we should explicitly handle the NotCoordinatorException case here. For the other cases, would it be possible to re-throw the exception?
It would be great if you could also add a few unit tests for this.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -747,4 +749,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
+ }
Review Comment:
Let's please add a unit test for this method.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -309,6 +311,21 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param context The request context.
+ * @param request The ListGroups request.
+ * @return A Result containing the ListGroupsResponseData response
+ */
+ public ListGroupsResponseData listGroups(
+ RequestContext context,
+ ListGroupsRequestData request,
+ long committedOffset
+ ) throws ApiException {
Review Comment:
I wonder if we should simplify the signature of this method. How about taking a list of states and returning a list of ListedGroup? The full request and the context are not really necessary in my opinion in this case.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
+ */
+ public ListGroupsResponseData listGroups(ListGroupsRequestData request, long committedOffset) {
+ Stream<Group> groupStream = groups.values(committedOffset).stream();
+ List<String> statesFilter = request.statesFilter();
+ if (!statesFilter.isEmpty()) {
+ groupStream = groupStream.filter(group -> statesFilter.contains(group.stateAsString()));
+ }
+ return new ListGroupsResponseData().setGroups(groupStream.map(Group::asListedGroup).collect(Collectors.toList()));
Review Comment:
In the ConsumerGroup case, the state is stored in a timeline data structure. Hence, we need to pass the `committedOffset` to `stateAsString` and `asListedGroup` in order to stay consistent.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
Review Comment:
nit: Remove `typesFilter`. `statesFilter` -> `states`.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -428,9 +430,44 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (TopicPartition tp : runtime.partitions()) {
+ futures.add(runtime.scheduleReadOperation(
+ "list_groups",
+ tp,
+ (coordinator, lastCommittedOffset) -> coordinator.listGroups(context, request, lastCommittedOffset)
+ ).exceptionally(exception -> {
+ if (!(exception instanceof KafkaException)) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, exception.getMessage());
+ }
+ return new ListGroupsResponseData()
+ .setErrorCode(Errors.forException(exception).code());
+ }));
+ }
+ CompletableFuture<ListGroupsResponseData> responseFuture = new CompletableFuture<>();
+ List<ListGroupsResponseData.ListedGroup> listedGroups = new ArrayList<>();
+ futures.forEach(CompletableFuture::join);
+ for (CompletableFuture<ListGroupsResponseData> future : futures) {
+ try {
+ ListGroupsResponseData data = future.get();
+ if (data.errorCode() != Errors.NONE.code()) {
+ responseFuture.complete(data);
+ return responseFuture;
+ }
+ listedGroups.addAll(future.get().groups());
+ } catch (InterruptedException | ExecutionException e) {
+ log.error("ListGroups request {} hit an unexpected exception: {}",
+ request, e.getMessage());
+ if (!responseFuture.isDone()) {
+ responseFuture.complete(new ListGroupsResponseData()
+ .setErrorCode(Errors.UNKNOWN_SERVER_ERROR.code()));
+ return responseFuture;
+ }
+ }
+ }
Review Comment:
Have you thought about writing an helper method to turn a list of CompletableFutures to a CompletableFuture containing the list of the result? That would be a nice building block that we could put in `FutureUtils`. If any of the CompletableFutures would fail, the resulting CompletableFuture would be failed as well with the same error. This would simplify the code here?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
+ */
+ public ListGroupsResponseData listGroups(ListGroupsRequestData request, long committedOffset) {
Review Comment:
Let's also add a unit test for this one.
--
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 #14271: KAFKA-14503: Implement ListGroups
Posted by "dajac (via GitHub)" <gi...@apache.org>.
dajac commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1302279096
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
Review Comment:
This seems to be inefficient because the coordinator may not be responsible for all the partitions. I thought that we could use `CoordinatorRuntime#partitions` to get the list of registered partitions. Have you considered this?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorShard.java:
##########
@@ -272,6 +274,21 @@ public CoordinatorResult<OffsetCommitResponseData, Record> commitOffset(
return offsetMetadataManager.commitOffset(context, request);
}
+ /**
+ * Handles a ListGroups request.
+ *
+ * @param context The request context.
+ * @param request The ListGroups request.
+ *
+ * @return A Result containing the ListGroupsResponseData response
+ */
+ public ListGroupsResponseData listGroups(
+ RequestContext context,
+ ListGroupsRequestData request
Review Comment:
nit: This should be indented with four spaces.
##########
clients/src/main/resources/common/message/ListGroupsRequest.json:
##########
@@ -23,11 +23,15 @@
// Version 3 is the first flexible version.
//
// Version 4 adds the StatesFilter field (KIP-518).
- "validVersions": "0-4",
+ //
+ // Version 5 adds the TypesFilter field (KIP-848).
+ "validVersions": "0-5",
"flexibleVersions": "3+",
"fields": [
{ "name": "StatesFilter", "type": "[]string", "versions": "4+",
"about": "The states of the groups we want to list. If empty all groups are returned with their state."
- }
+ },
+ { "name": "TypesFilter", "type": "[]string", "versions": "5+",
+ "about": "The types of the groups we want to list. If empty all groups are returned" }
Review Comment:
I would rather prefer to do this in a second PR because this change impacts both the new and the old group coordinators. Would it be possible?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
Review Comment:
nit: This should be indented with four spaces. I have seen this in other places in the code but I am not going to comment them all. I let you have a look.
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
Review Comment:
Don't we need to also implement this for `GenericGroup`?
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
+ futures.add(runtime.scheduleReadOperation("list_groups",
+ new TopicPartition(Topic.GROUP_METADATA_TOPIC_NAME, i),
+ (coordinator, __) -> coordinator.listGroups(context, request)
Review Comment:
We need to use the second parameters `__` as pass it down to `listGroups`. For the context, the second parameter is the last committed offsets. We should list the groups based on it. Otherwise, we would return uncommitted changes.
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1692157804
> > @dajac hi, could you pls help me see the check error log. I don't quite understand the root cause and what should be fixed myself. The check error log is as followed
> > > coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:129:1: Class Data Abstraction Coupling is 26 (max allowed is 25) classes [CompletableFuture, ConsumerGroup, ConsumerGroupHeartbeatResponseData, ConsumerGroupHeartbeatResponseData.Assignment, ConsumerGroupHeartbeatResponseData.TopicPartitions, ConsumerGroupMember.Builder, CoordinatorResult, CurrentAssignmentBuilder, FencedMemberEpochException, GenericGroup, GenericGroupMember, GroupIdNotFoundException, GroupMaxSizeReachedException, HeartbeatResponseData, InvalidRequestException, JoinGroupRequestProtocol, JoinGroupRequestProtocolCollection, JoinGroupResponseData, ListGroupsResponseData, SyncGroupResponseData, TargetAssignmentBuilder, TimelineHashMap, TimelineHashSet, UnknownMemberIdException, UnknownServerException, UnsupportedAssignorException].
>
> @zhaohaidao It looks like we have hit the maximum number of imports. We could consider adding an exception for this. Have you tried this?
Thanks for your suggestion. this err has been fixed
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304506722
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
Review Comment:
done
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1314217786
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupMetadataManager.java:
##########
@@ -420,6 +423,18 @@ public Group group(String groupId, long committedOffset) throws GroupIdNotFoundE
return group;
}
+ /**
+ * @return The GenericGroup List filtered by statesFilter or typesFilter.
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
[GitHub] [kafka] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304474195
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java:
##########
@@ -426,9 +429,43 @@ public CompletableFuture<ListGroupsResponseData> listGroups(
return FutureUtils.failedFuture(Errors.COORDINATOR_NOT_AVAILABLE.exception());
}
- return FutureUtils.failedFuture(Errors.UNSUPPORTED_VERSION.exception(
- "This API is not implemented yet."
- ));
+ List<CompletableFuture<ListGroupsResponseData>> futures = new java.util.ArrayList<>(Collections.emptyList());
+ for (int i = 0; i < numPartitions; i++) {
Review Comment:
Thanks for the reminder, I tried to find it but didn't know it, so I chose an inefficient implementation.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1304540159
##########
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/consumer/ConsumerGroup.java:
##########
@@ -739,4 +741,15 @@ private static Integer decValue(String key, Integer value) {
private static Integer incValue(String key, Integer value) {
return value == null ? 1 : value + 1;
}
+
+ /**
+ * @return the group formatted as a list group response.
+ */
+ public ListGroupsResponseData.ListedGroup asListedGroup() {
+ return new ListGroupsResponseData.ListedGroup()
+ .setGroupId(groupId)
+ .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
+ .setGroupState(state.toString());
Review Comment:
Thank you for the suggestion. I have fixed all the indented issues introduced by this pr.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1324884926
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -431,14 +434,17 @@ public void rollback() {
snapshotRegistry.revertToSnapshot(lastCommittedOffset);
}
+ public void getOrCreateSnapshot() {
+ snapshotRegistry.getOrCreateSnapshot(lastCommittedOffset);
+ }
Review Comment:
This will cause testListGroups to fail because groupMetadataManager.listGroups always uses lastCommittedOffset to query. Putting getOrCreateSnapshot(lastWrittenOffset) at the beginning of the replay function will lead to the following result: the largest epoch in the snapshot is always 1 less than lastCommittedOffset. In this way, listGroups cannot find the expected results.
Put getOrCreateSnapshot(lastWrittenOffset) at the end of the function, especially after lastWrittenOffset+1, all cases can pass, which is consistent with the semantics of CoordinatorRuntime, that is, always update the memory first, then update lastWrittenOffset, and use the latest lastWrittenOffset Update snapshot.
WDYT?
--
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] zhaohaidao commented on pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on PR #14271:
URL: https://github.com/apache/kafka/pull/14271#issuecomment-1718110509
> @zhaohaidao Thanks for all the work you are doing on this. I really appreciate it. I have created a few jiras for the next steps:
>
> * https://issues.apache.org/jira/browse/KAFKA-15460
> * https://issues.apache.org/jira/browse/KAFKA-15461
> * https://issues.apache.org/jira/browse/KAFKA-15462
>
> I have assigned all of them to you for now as you were interested in pursuing this work. Feel free to unassign them if you don't want or plan to tackle them.
>
> From a priority perspective, tacking https://issues.apache.org/jira/browse/KAFKA-14504 before the above mentioned jiras would be really helpful as it will allow us to start end-to-end testing when we have all the existing APIs in place.
Thank you for your suggestion. I will give priority to 14504 and deal with other tasks one after another.
--
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] zhaohaidao commented on a diff in pull request #14271: KAFKA-14503: Implement ListGroups
Posted by "zhaohaidao (via GitHub)" <gi...@apache.org>.
zhaohaidao commented on code in PR #14271:
URL: https://github.com/apache/kafka/pull/14271#discussion_r1323252098
##########
group-coordinator/src/test/java/org/apache/kafka/coordinator/group/GroupMetadataManagerTest.java:
##########
@@ -8574,6 +8597,36 @@ public void testHeartbeatDuringRebalanceCausesRebalanceInProgress() throws Excep
assertEquals(Errors.REBALANCE_IN_PROGRESS.code(), heartbeatResponse.errorCode());
}
+ @Test
+ public void testListGroups() {
+ String genericGroupId = "generic-group-id";
+ String consumerGroupId = "consumer-group-id";
+ GroupMetadataManagerTestContext context = new GroupMetadataManagerTestContext.Builder().build();
+ context.updateLastWrittenOffset(context.lastWrittenOffset);
+ GenericGroup genericGroup = context.createGenericGroup(genericGroupId);
+ ConsumerGroup consumerGroup = context.createConsumerGroup(consumerGroupId);
+ context.updateLastWrittenOffset(context.lastWrittenOffset + 2);
Review Comment:
I updated this test case based on my own understanding. Please help to see if it meets expectations.
--
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