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-unsubscr...@kafka.apache.org
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
