{"record":{"id":"3534f82922059b44","repo":"apache/beam","slug":"error-decoding-input-stream-with-coder-coder-in-step-step","errorCode":null,"errorMessage":"Error decoding input stream with coder {coder} in step {step}","messagePattern":"Error decoding input stream with coder (.+?) in step (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/bundle_processor.py","lineNumber":232,"sourceCode":"      self.started = True\n\n  def process(self, windowed_value: windowed_value.WindowedValue) -> None:\n    self.output(windowed_value)\n\n  def process_encoded(self, encoded_windowed_values: bytes) -> None:\n    input_stream = coder_impl.create_InputStream(encoded_windowed_values)\n    while input_stream.size() > 0:\n      with self.splitting_lock:\n        if self.index == self.stop - 1:\n          return\n        self.index += 1\n      try:\n        decoded_value = self.windowed_coder_impl.decode_from_stream(\n            input_stream, True)\n      except Exception as exn:\n        coder = str(self.windowed_coder)\n        step = self.name_context.step_name\n        raise ValueError(\n            f\"Error decoding input stream with coder {coder} in step {step}\"\n        ) from exn\n      self.output(decoded_value)\n\n  def monitoring_infos(\n      self, transform_id: str, tag_to_pcollection_id: dict[str, str]\n  ) -> dict[frozenset, metrics_pb2.MonitoringInfo]:\n    all_monitoring_infos = super().monitoring_infos(\n        transform_id, tag_to_pcollection_id)\n    read_progress_info = monitoring_infos.int64_counter(\n        monitoring_infos.DATA_CHANNEL_READ_INDEX,\n        self.index,\n        ptransform=transform_id)\n    all_monitoring_infos[monitoring_infos.to_key(\n        read_progress_info)] = read_progress_info\n    return all_monitoring_infos\n\n  # TODO(https://github.com/apache/beam/issues/19737): typing not compatible","sourceCodeStart":214,"sourceCodeEnd":250,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/bundle_processor.py#L214-L250","documentation":"The SDK worker failed to decode an incoming data stream using the step's windowed coder, so bundle_processor wraps the original exception in a ValueError identifying the coder and step. It's a wrap-and-annotate of an underlying decode failure (corrupt bytes, coder mismatch, schema change).","triggerScenarios":"Data in an incoming gRPC data stream cannot be decoded by self.windowed_coder_impl.decode_from_stream in DataChannel.OperationInbox.process_encoded — e.g. bytes written with a different coder than declared, truncated/corrupt data, or a changed Avro/Proto/Row schema.","commonSituations":"Updating a schema (adding fields with incompatible encoding) between job submission and worker; elements serialized by a different Beam/interpreter version; manual coder mismatch when using custom coders.","solutions":["Read the chained 'from exn' cause to find the real decode error and fix it at the source","Ensure the coder registered for the step matches the one used to write the data (re-run pipeline stages consistently)","Pin matching Apache Beam versions across submission and worker environments","If using custom coders, add a version/compatibility marker or tolerant decoding"],"exampleFix":"// before\ntry:\n    result = coder.decode(record)\n// after\ntry:\n    result = coder.decode(record)\nexcept Exception as e:\n    logging.warning('Skipping undecodable record with coder %s: %s', coder, e)\n    return  # or re-encode with the correct coder","handlingStrategy":"try-catch","validationCode":"import apache_beam as beam\nsub_v = beam.version.__version__\n# ensure submission SDK version matches the container image version used by the runner\nassert sub_v == os.environ.get('BEAM_SDK_VERSION_IN_IMAGE'), 'Beam version mismatch'\n","typeGuard":null,"tryCatchPattern":"try:\n    result = bundle_processor.process(encoded_stream)\nexcept ValueError as e:\n    if 'Error decoding input stream with coder' in str(e):\n        logging.error('Decode failed: %s — inspect cause', e.__cause__)\n        # route record to a dead-letter sink instead of failing the bundle\n        dead_letter(record, cause=e.__cause__)\n    else:\n        raise","preventionTips":["Always inspect the __cause__ chained exception for the root decode failure","Keep Beam and schema versions identical between submission and workers","Version your schemas; only make backward-compatible encoding changes","Test custom coders round-trip (encode->decode) in unit tests"],"tags":["apache-beam","python","decoding","coder"],"backgroundTag":"protobuf-unmarshal-failed","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T16:17:12.679Z"}