@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())
            ));
    }
See {@link GroupCoordinator#listGroups(RequestContext, ListGroupsRequestData)}.
group-coordinator/src/main/java/org/apache/kafka/coordinator/group/GroupCoordinatorService.java