def _send_group_coordinator_request(self):
        """Discover the current coordinator for the group.

        Returns:
            Future: resolves to the node id of the coordinator
        """
        node_id = self._client.least_loaded_node()
        if node_id is None or self._client.cluster.is_bootstrap(node_id):
            return Future().failure(Errors.NoBrokersAvailable())

        elif not self._client.ready(node_id, metadata_priority=False):
            e = Errors.NodeNotReadyError(node_id)
            return Future().failure(e)

        version = self._client.api_version(FindCoordinatorRequest, max_version=2)
        if version == 0:
            request = FindCoordinatorRequest[version](self.group_id)
        else:
            request = FindCoordinatorRequest[version](self.group_id, 0)
        log.debug("Sending group coordinator request for group %s to broker %s: %s",
                  self.group_id, node_id, request)
        future = Future()
        _f = self._client.send(node_id, request)
        _f.add_callback(self._handle_group_coordinator_response, future)
        _f.add_errback(self._failed_request, node_id, request, future)
        return future
Discover the current coordinator for the group.

        Returns:
            Future: resolves to the node id of the coordinator
        
kafka/coordinator/base.py