def retry(self, retry_message=''):
    """Forces a currently running asynchronous pipeline to retry.

    Note this may not be called by synchronous or generator pipelines. Those
    must instead raise the 'Retry' exception during execution.

    Args:
      retry_message: Optional message explaining why the retry happened.

    Returns:
      True if the Pipeline should be retried, False if it cannot be cancelled
      mid-flight for some reason.
    """
    if not self.async:
      raise UnexpectedPipelineError(
          'May only call retry() method for asynchronous pipelines.')
    if self.try_cancel():
      self._context.transition_retry(self._pipeline_key, retry_message)
      return True
    else:
      return False
Forces a currently running asynchronous pipeline to retry.

    Note this may not be called by synchronous or generator pipelines. Those
    must instead raise the 'Retry' exception during execution.

    Args:
      retry_message: Optional message explaining why the retry happened.

    Returns:
      True if the Pipeline should be retried, False if it cannot be cancelled
      mid-flight for some reason.