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.CoordinatorNotAvailableError(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()
        ]
        version = self._client.api_version(JoinGroupRequest, max_version=4)
        if version == 0:
            request = JoinGroupRequest[version](
                self.group_id,
                self.config['session_timeout_ms'],
                self._generation.member_id,
                self.protocol_type(),
                member_metadata)
        else:
            request = JoinGroupRequest[version](
                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
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
        
kafka/coordinator/base.py