apache/beam · error · RuntimeError

Unexpected data type

Error message

Unexpected data type: %s

What it means

During execute_executable_stage, the runner iterates over the elements returned by the SDK worker and only knows how to handle Elements.Data messages (plain data outputs). Any other element type (e.g. timers or other control payloads) is rejected with 'Unexpected data type'. The trivial runner simply cannot process those response kinds.

Solutions

  1. Avoid timers/stateful transforms when using TrivialRunner; use DirectRunner instead
  2. Upgrade apache_beam so trivial_runner handles the new element types
  3. Restructure the pipeline to emit only regular data outputs under the trivial runner
  4. File/patch the runner to handle the specific Elements oneof variant

Example fix

// before
pipeline.run(runner=TrivialRunner())  # pipeline uses timers
// after
pipeline.run(runner=DirectRunner())  # timers supported
Defensive patterns

Strategy: try-catch

Validate before calling

def uses_timers_or_state(pipeline):
    for t in pipeline.proto.components.transforms.values():
        if 'timer' in (t.spec.urn or '').lower() or 'state' in (t.spec.urn or '').lower():
            return True
    return False
# if True, don't use TrivialRunner

Try / catch

try:
    pipeline.run(runner=TrivialRunner()).wait_until_finish()
except RuntimeError as e:
    if 'Unexpected data type' in str(e):
        logging.warning('TrivialRunner cannot handle this element type; using DirectRunner')
        pipeline.run(runner=DirectRunner())
    else:
        raise

Prevention

When it happens

Trigger: Running a pipeline under TrivialRunner where the executed bundle emits non-data responses — most commonly timer firings (Elements.Timer) or other control messages from the Fn API process bundle.

Common situations: Pipelines using timers/stateful DoFns tested with the trivial runner; SDK worker returning control payloads the trivial runner doesn't model; version drift between SDK producing new element types and an older trivial_runner.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/trivial_runner.py:234

    process_bundle_request = beam_fn_api_pb2.InstructionRequest(
        instruction_id=process_bundle_id,
        process_bundle=beam_fn_api_pb2.ProcessBundleRequest(
            process_bundle_descriptor_id=process_bundle_descriptor.id))
    result_future = execution_state.worker_handler.control_conn.push(
        process_bundle_request)

    # Read the results off the data channel.
    # Note that if there are multiple outputs, we may get them in any order,
    # possibly interleaved.
    for output in execution_state.worker_handler.data_conn.input_elements(
        process_bundle_id, list(output_ops_to_pcoll.keys())):
      if isinstance(output, beam_fn_api_pb2.Elements.Data):
        # Adds the output to the appropriate PCollection.
        execution_state.set_pcollection_contents(
            output_ops_to_pcoll[output.transform_id], [output.data])
      else:
        # E.g. timers to set.
        raise RuntimeError("Unexpected data type: %s" % output)

    # Ensure the operation completed successfully.
    # This result contains things like metrics and continuation tokens as well.
    result = result_future.get()
    if result.error:
      raise RuntimeError(result.error)
    if result.process_bundle.residual_roots:
      # We would need to re-schedule execution of this bundle with this data.
      raise NotImplementedError('SDF continuation')
    if result.process_bundle.requires_finalization:
      # We would need to invoke the finalization callback, on a best effort
      # basis, *after* the outputs are durably committed.
      raise NotImplementedError('finalization')
    if result.process_bundle.elements.data:
      # These should be processed just like outputs from the data channel.
      raise NotImplementedError('control-channel data')
    if result.process_bundle.elements.timers:
      # These should be processed just like outputs from the data channel.

View on GitHub (pinned to 12126d8942)