'Join the group and return the assignment for the next generation.
This function handles both JoinGroup and SyncGroup, delegating to
:meth:`._perform_assignment` if elected leader by the coordinator.
Returns:
Future: resolves to the encoded-bytes assignment returned from the
group leader'

def _send_join_group_request(self):

    if self.coordinator_unknown():
        e = Errors.GroupCoordinatorNotAvailableError(self.coordinator_id)
        return Future().failure(e)
    elif (not self._client.ready(self.coordinator_id, metadata_priority=False)):
        e = Errors.NodeNotReadyError(self.coordinator_id)
        return Future().failure(e)
    log.info('(Re-)joining    group    %s', self.group_id)
    request = JoinGroupRequest[0](self.group_id, self.config['session_timeout_ms'], self.member_id, self.protocol_type(), [(protocol, (metadata if isinstance(metadata, bytes) else metadata.encode())) for (protocol, metadata) in self.group_protocols()])
    log.debug('Sending    JoinGroup    (%s)    to    coordinator    %s', request, self.coordinator_id)
    future = Future()
    _f = self._client.send(self.coordinator_id, request)
    _f.add_callback(self._handle_join_group_response, future, time.time())
    _f.add_errback(self._failed_request, self.coordinator_id, request, future)
    return future
