apache/beam · error · ValueError

Unexpected input element type

Error message

Unexpected input element type %s

What it means

input_elements yields Data/Timers elements and consumes is_last markers per transform. If an element from the queue is neither a data/timer element with a recognized shape, a ValueError 'Unexpected input element type' is raised, guarding against corrupt or unexpected protobuf Elements payloads.

Solutions

1. Ensure runner and SDK harness use the same Apache Beam version (pin versions across the deployment).
2. Check for accidental downgrades/upgrade of the beam package at runtime in the worker image.
3. Inspect the payload producer (custom sources/transforms) that enqueues elements into the data plane.
4. File/inspect a Beam issue if reproducible with stock components; this should be unreachable in normal use.

Example fix

// before
# runner: apache-beam==2.50.0, worker image: apache-beam==2.58.0
// after
# pin both to the same version, e.g.
# runner: apache-beam==2.58.0, worker image: apache-beam==2.58.0

# verify version parity before submitting
import apache_beam as beam
assert runner_beam_version == beam.__version__, 'runner/SDK Beam version mismatch'

When it happens

Trigger: An element popped from the receiving queue whose type does not match beam_fn_api_pb2.Elements.Data/Timers expectations in the element handling branches (e.g. a malformed or future-version protobuf message).

Common situations: SDK/runner version skew where one side emits newer protobuf fields the other cannot interpret; corrupted internal queue contents from a buggy custom transform; mixing data plane ports/clients.

Related errors

• Cannot interpret a request received over control channel…
• Encountered unknown AtomicType…
• Cannot convert from nanoseconds to microseconds because…
• cannot encode a null ByteString
• Cannot provide because is not a subclass of

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/worker/data_plane.py:625

                instruction_id,
                current_time - start_time)
            next_waiting_log_time = current_time + log_interval_sec
        else:
          start_time = time.time()
          next_waiting_log_time = start_time + log_interval_sec
          if isinstance(element, beam_fn_api_pb2.Elements.Timers):
            if element.is_last:
              done_inputs.add((element.transform_id, element.timer_family_id))
            else:
              yield element
          elif isinstance(element, beam_fn_api_pb2.Elements.Data):
            if element.is_last:
              done_inputs.add(element.transform_id)
            else:
              assert element.transform_id not in done_inputs
              yield element
          else:
            raise ValueError('Unexpected input element type %s' % type(element))
    finally:
      # Instruction_ids are not reusable so Clean queue once we are done with
      #  an instruction_id
      self._clean_receiving_queue(instruction_id)

  def output_stream(self, instruction_id, transform_id):
    # type: (str, str) -> ClosableOutputStream
    def add_to_send_queue(data):
      # type: (bytes) -> None
      if data:
        elem = beam_fn_api_pb2.Elements.Data(
            instruction_id=instruction_id, transform_id=transform_id, data=data)
        self._enqueue_to_send(elem)

    def close_callback(data):
      # type: (bytes) -> None
      add_to_send_queue(data)
      # End of stream marker.

View on GitHub (pinned to 12126d8942)