def _send_join_group_request(self):
        """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
        """
        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)

        # send a join group request to the coordinator
        log.info("(Re-)joining group %s", self.group_id)
        member_metadata = [
            (protocol, metadata if isinstance(metadata, bytes) else metadata.encode())
            for protocol, metadata in self.group_protocols()
        ]
        if self.config['api_version'] < (0, 9):
            raise Errors.KafkaError('JoinGroupRequest api requires 0.9+ brokers')
        elif (0, 9) <= self.config['api_version'] < (0, 10, 1):
            request = JoinGroupRequest[0](
                self.group_id,
                self.config['session_timeout_ms'],
                self._generation.member_id,
                self.protocol_type(),
                member_metadata)
        elif (0, 10, 1) <= self.config['api_version'] < (0, 11, 0):
            request = JoinGroupRequest[1](
                self.group_id,
                self.config['session_timeout_ms'],
                self.config['max_poll_interval_ms'],
                self._generation.member_id,
                self.protocol_type(),
                member_metadata)
        else:
            request = JoinGroupRequest[2](
                self.group_id,
                self.config['session_timeout_ms'],
                self.config['max_poll_interval_ms'],
                self._generation.member_id,
                self.protocol_type(),
                member_metadata)

        # create the request for the coordinator
        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