docs: 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