apache/beam · error · PermanentException

Skip retrying because we caught exception

Error message

Skip retrying because we caught exception:%s, but the stream is not seekable.

What it means

stage_file_with_retry retries staging a file by rewinding (seek 0) the stream between attempts. If a retryable exception occurs but the stream is not seekable, the cursor cannot be reset, so the code gives up with a PermanentException embedding the original exception text.

Solutions

1. Buffer the non-seekable stream into a local temp file or BytesIO and stage that
2. Provide a seekable file object (open() on a real path) instead of a pipe
3. Pre-size and fully materialize the data before calling the staging API

Example fix

// before
stage_file_with_retry(sys.stdin.buffer, dest)
// after
import tempfile, shutil
with tempfile.TemporaryFile() as tmp:
    shutil.copyfileobj(sys.stdin.buffer, tmp)
    tmp.seek(0)
    stage_file_with_retry(tmp, dest)

if not stream.seekable():
    buf = io.BytesIO(stream.read())
    buf.seek(0)
    stream = buf

def is_stageable(obj):
    return hasattr(obj, 'seek') and obj.seekable() or isinstance(obj, (str, bytes)) and os.path.exists(obj)

try:
    stage_file_with_retry(stream, dest)
except retry.PermanentException as ex:
    logger.error('staging aborted (non-seekable stream): %s', ex)
    raise

When it happens

Trigger: Passing a non-seekable stream (e.g. sys.stdin, a socket/file-like object without seek) to stage_file_with_retry and the upload fails once, forcing an un-retryable abort.

Common situations: Piping data from stdin/pipe into GCS staging; custom file-like wrappers lacking seek(); network failure mid-upload of streamed content.

Related errors

• Skip retrying because type %sstream_or_path is unsupported.
• cache_root GCS bucket path is invalid.
• Caught retentionPolicyNotMet error while rewriting to a…
• Could not upload to GCS path
• Could not upload to GCS path

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/845afd74a6c4c5ca. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/dataflow/internal/apiclient.py:753

      total_size=None):

    if isinstance(stream_or_path, str):
      path = stream_or_path
      with open(path, 'rb') as stream:
        self.stage_file(
            gcs_or_local_path, file_name, stream, mime_type, total_size)
    elif isinstance(stream_or_path, io.IOBase):
      stream = stream_or_path
      try:
        self.stage_file(
            gcs_or_local_path, file_name, stream, mime_type, total_size)
      except Exception as exn:
        if stream.seekable():
          # reset cursor for possible retrying
          stream.seek(0)
          raise exn
        else:
          raise retry.PermanentException(
              "Skip retrying because we caught exception:" +
              ''.join(traceback.format_exception_only(exn.__class__, exn)) +
              ', but the stream is not seekable.')
    else:
      raise retry.PermanentException(
          "Skip retrying because type " + str(type(stream_or_path)) +
          "stream_or_path is unsupported.")

  @retry.no_retries  # Using no_retries marks this as an integration point.
  def create_job(self, job: Job):
    """Creates job description. May stage and/or submit for remote execution."""
    self.create_job_description(job)

    # Stage and submit the job when necessary
    dataflow_job_file = job.options.view_as(DebugOptions).dataflow_job_file
    template_location = (
        job.options.view_as(GoogleCloudOptions).template_location)

View on GitHub (pinned to 12126d8942)