apache/beam · error · ValueError

Error decoding input stream with coder {coder} in step {step

Error message

Error decoding input stream with coder {coder} in step {step}

What it means

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).

Source

Thrown at sdks/python/apache_beam/runners/worker/bundle_processor.py:232

      self.started = True

  def process(self, windowed_value: windowed_value.WindowedValue) -> None:
    self.output(windowed_value)

  def process_encoded(self, encoded_windowed_values: bytes) -> None:
    input_stream = coder_impl.create_InputStream(encoded_windowed_values)
    while input_stream.size() > 0:
      with self.splitting_lock:
        if self.index == self.stop - 1:
          return
        self.index += 1
      try:
        decoded_value = self.windowed_coder_impl.decode_from_stream(
            input_stream, True)
      except Exception as exn:
        coder = str(self.windowed_coder)
        step = self.name_context.step_name
        raise ValueError(
            f"Error decoding input stream with coder {coder} in step {step}"
        ) from exn
      self.output(decoded_value)

  def monitoring_infos(
      self, transform_id: str, tag_to_pcollection_id: dict[str, str]
  ) -> dict[frozenset, metrics_pb2.MonitoringInfo]:
    all_monitoring_infos = super().monitoring_infos(
        transform_id, tag_to_pcollection_id)
    read_progress_info = monitoring_infos.int64_counter(
        monitoring_infos.DATA_CHANNEL_READ_INDEX,
        self.index,
        ptransform=transform_id)
    all_monitoring_infos[monitoring_infos.to_key(
        read_progress_info)] = read_progress_info
    return all_monitoring_infos

  # TODO(https://github.com/apache/beam/issues/19737): typing not compatible

View on GitHub (pinned to 12126d8942)

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

Example fix

// before
try:
    result = coder.decode(record)
// after
try:
    result = coder.decode(record)
except Exception as e:
    logging.warning('Skipping undecodable record with coder %s: %s', coder, e)
    return  # or re-encode with the correct coder

Defensive patterns

Strategy: try-catch

Validate before calling

import apache_beam as beam
sub_v = beam.version.__version__
# ensure submission SDK version matches the container image version used by the runner
assert sub_v == os.environ.get('BEAM_SDK_VERSION_IN_IMAGE'), 'Beam version mismatch'

Try / catch

try:
    result = bundle_processor.process(encoded_stream)
except ValueError as e:
    if 'Error decoding input stream with coder' in str(e):
        logging.error('Decode failed: %s — inspect cause', e.__cause__)
        # route record to a dead-letter sink instead of failing the bundle
        dead_letter(record, cause=e.__cause__)
    else:
        raise

Prevention

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

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: "cannot parse invalid wire-format data", "cannot unmarshal", "failed unmarshalling": protobuf unmarshal errors explained — this error's family across 10 libraries.

Related errors

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