@Override
    public CompletableFuture<ListGroupsResponseData> listGroups(
        RequestContext context,
        ListGroupsRequestData request
    ) {
        if (!isActive.get()) {
            return CompletableFuture.completedFuture(new ListGroupsResponseData()
                .setErrorCode(Errors.COORDINATOR_NOT_AVAILABLE.code())
            );
        }

        final List<CompletableFuture<List<ListGroupsResponseData.ListedGroup>>> futures = FutureUtils.mapExceptionally(
            runtime.scheduleReadAllOperation(
                "list-groups",
                (coordinator, lastCommittedOffset) -> coordinator.listGroups(
                    request.statesFilter(),
                    request.typesFilter(),
                    lastCommittedOffset
                )
            ),
            exception -> {
                exception = Errors.maybeUnwrapException(exception);
                if (exception instanceof NotCoordinatorException) {
                    return Collections.emptyList();
                } else {
                    throw new CompletionException(exception);
                }
            }
        );

        return FutureUtils
            .combineFutures(futures, ArrayList::new, List::addAll)
            .thenApply(groups -> new ListGroupsResponseData().setGroups(groups))
            .exceptionally(exception -> handleOperationException(
                "list-groups",
                request,
                exception,
                (error, __) -> new ListGroupsResponseData().setErrorCode(error.code())
            ));
    }
@Test
    public void testListGroups() throws ExecutionException, InterruptedException, TimeoutException {
        CoordinatorRuntime<GroupCoordinatorShard, CoordinatorRecord> runtime = mockRuntime();
        GroupCoordinatorService service = new GroupCoordinatorService(
            new LogContext(),
            createConfig(),
            runtime,
            new GroupCoordinatorMetrics(),
            createConfigManager()
        );
        service.startup(() -> 3);

        List<ListGroupsResponseData.ListedGroup> expectedResults = Arrays.asList(
            new ListGroupsResponseData.ListedGroup()
                .setGroupId("group0")
                .setProtocolType("protocol1")
                .setGroupState("Stable")
                .setGroupType("classic"),
            new ListGroupsResponseData.ListedGroup()
                .setGroupId("group1")
                .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                .setGroupState("Empty")
                .setGroupType("consumer"),
            new ListGroupsResponseData.ListedGroup()
                .setGroupId("group2")
                .setProtocolType(ConsumerProtocol.PROTOCOL_TYPE)
                .setGroupState("Dead")
                .setGroupType("consumer")
        );

        when(runtime.scheduleReadAllOperation(
            ArgumentMatchers.eq("list-groups"),
            ArgumentMatchers.any()
        )).thenReturn(Arrays.asList(
            CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(0))),
            CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(1))),
            CompletableFuture.completedFuture(Collections.singletonList(expectedResults.get(2)))
        ));

        CompletableFuture<ListGroupsResponseData> responseFuture = service.listGroups(
            requestContext(ApiKeys.LIST_GROUPS),
            new ListGroupsRequestData()
        );

        assertEquals(expectedResults, responseFuture.get(5, TimeUnit.SECONDS).groups());
    }