def _describe_consumer_groups_send_request(self, group_id, group_coordinator_id, include_authorized_operations=False):
        """Send a DescribeGroupsRequest to the group's coordinator.

        Arguments:
            group_id: The group name as a string
            group_coordinator_id: The node_id of the groups' coordinator broker.

        Returns:
            A message future.
        """
        version = self._client.api_version(DescribeGroupsRequest, max_version=3)
        if version <= 2:
            if include_authorized_operations:
                raise IncompatibleBrokerVersion(
                    "include_authorized_operations requests "
                    "DescribeGroupsRequest >= v3, which is not "
                    "supported by Kafka {}".format(version)
                )
            # Note: KAFKA-6788 A potential optimization is to group the
            # request per coordinator and send one request with a list of
            # all consumer groups. Java still hasn't implemented this
            # because the error checking is hard to get right when some
            # groups error and others don't.
            request = DescribeGroupsRequest[version](groups=(group_id,))
        elif version <= 3:
            request = DescribeGroupsRequest[version](
                groups=(group_id,),
                include_authorized_operations=include_authorized_operations
            )
        else:
            raise NotImplementedError(
                "Support for DescribeGroupsRequest_v{} has not yet been added to KafkaAdminClient."
                .format(version))
        return self._send_request_to_node(group_coordinator_id, request)
Send a DescribeGroupsRequest to the group's coordinator.

        Arguments:
            group_id: The group name as a string
            group_coordinator_id: The node_id of the groups' coordinator broker.

        Returns:
            A message future.
        
kafka/admin/client.py