{"record":{"id":"845afd74a6c4c5ca","repo":"apache/beam","slug":"skip-retrying-because-we-caught-exception-s-but-the-stream","errorCode":null,"errorMessage":"Skip retrying because we caught exception:%s, but the stream is not seekable.","messagePattern":"Skip retrying because we caught exception:(.+?), but the stream is not seekable\\.","errorType":"exception","errorClass":"PermanentException","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/dataflow/internal/apiclient.py","lineNumber":753,"sourceCode":"      total_size=None):\n\n    if isinstance(stream_or_path, str):\n      path = stream_or_path\n      with open(path, 'rb') as stream:\n        self.stage_file(\n            gcs_or_local_path, file_name, stream, mime_type, total_size)\n    elif isinstance(stream_or_path, io.IOBase):\n      stream = stream_or_path\n      try:\n        self.stage_file(\n            gcs_or_local_path, file_name, stream, mime_type, total_size)\n      except Exception as exn:\n        if stream.seekable():\n          # reset cursor for possible retrying\n          stream.seek(0)\n          raise exn\n        else:\n          raise retry.PermanentException(\n              \"Skip retrying because we caught exception:\" +\n              ''.join(traceback.format_exception_only(exn.__class__, exn)) +\n              ', but the stream is not seekable.')\n    else:\n      raise retry.PermanentException(\n          \"Skip retrying because type \" + str(type(stream_or_path)) +\n          \"stream_or_path is unsupported.\")\n\n  @retry.no_retries  # Using no_retries marks this as an integration point.\n  def create_job(self, job: Job):\n    \"\"\"Creates job description. May stage and/or submit for remote execution.\"\"\"\n    self.create_job_description(job)\n\n    # Stage and submit the job when necessary\n    dataflow_job_file = job.options.view_as(DebugOptions).dataflow_job_file\n    template_location = (\n        job.options.view_as(GoogleCloudOptions).template_location)\n","sourceCodeStart":735,"sourceCodeEnd":771,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/dataflow/internal/apiclient.py#L735-L771","documentation":"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.","triggerScenarios":"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.","commonSituations":"Piping data from stdin/pipe into GCS staging; custom file-like wrappers lacking seek(); network failure mid-upload of streamed content.","solutions":["Buffer the non-seekable stream into a local temp file or BytesIO and stage that","Provide a seekable file object (open() on a real path) instead of a pipe","Pre-size and fully materialize the data before calling the staging API"],"exampleFix":"// before\nstage_file_with_retry(sys.stdin.buffer, dest)\n// after\nimport tempfile, shutil\nwith tempfile.TemporaryFile() as tmp:\n    shutil.copyfileobj(sys.stdin.buffer, tmp)\n    tmp.seek(0)\n    stage_file_with_retry(tmp, dest)","handlingStrategy":"fallback","validationCode":"if not stream.seekable():\n    buf = io.BytesIO(stream.read())\n    buf.seek(0)\n    stream = buf","typeGuard":"def is_stageable(obj):\n    return hasattr(obj, 'seek') and obj.seekable() or isinstance(obj, (str, bytes)) and os.path.exists(obj)","tryCatchPattern":"try:\n    stage_file_with_retry(stream, dest)\nexcept retry.PermanentException as ex:\n    logger.error('staging aborted (non-seekable stream): %s', ex)\n    raise","preventionTips":["Materialize streamed data into temp files before staging","Ensure custom file-like wrappers implement seek()","Prefer file paths over pipes for uploads"],"tags":["python","gcs","upload","retry","stream"],"backgroundTag":"retry-not-possible","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}