{"record":{"id":"f5d097768c912cb1","repo":"apache/beam","slug":"unexpected-data-type-s","errorCode":null,"errorMessage":"Unexpected data type: %s","messagePattern":"Unexpected data type: (.+?)","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/trivial_runner.py","lineNumber":234,"sourceCode":"    process_bundle_request = beam_fn_api_pb2.InstructionRequest(\n        instruction_id=process_bundle_id,\n        process_bundle=beam_fn_api_pb2.ProcessBundleRequest(\n            process_bundle_descriptor_id=process_bundle_descriptor.id))\n    result_future = execution_state.worker_handler.control_conn.push(\n        process_bundle_request)\n\n    # Read the results off the data channel.\n    # Note that if there are multiple outputs, we may get them in any order,\n    # possibly interleaved.\n    for output in execution_state.worker_handler.data_conn.input_elements(\n        process_bundle_id, list(output_ops_to_pcoll.keys())):\n      if isinstance(output, beam_fn_api_pb2.Elements.Data):\n        # Adds the output to the appropriate PCollection.\n        execution_state.set_pcollection_contents(\n            output_ops_to_pcoll[output.transform_id], [output.data])\n      else:\n        # E.g. timers to set.\n        raise RuntimeError(\"Unexpected data type: %s\" % output)\n\n    # Ensure the operation completed successfully.\n    # This result contains things like metrics and continuation tokens as well.\n    result = result_future.get()\n    if result.error:\n      raise RuntimeError(result.error)\n    if result.process_bundle.residual_roots:\n      # We would need to re-schedule execution of this bundle with this data.\n      raise NotImplementedError('SDF continuation')\n    if result.process_bundle.requires_finalization:\n      # We would need to invoke the finalization callback, on a best effort\n      # basis, *after* the outputs are durably committed.\n      raise NotImplementedError('finalization')\n    if result.process_bundle.elements.data:\n      # These should be processed just like outputs from the data channel.\n      raise NotImplementedError('control-channel data')\n    if result.process_bundle.elements.timers:\n      # These should be processed just like outputs from the data channel.","sourceCodeStart":216,"sourceCodeEnd":252,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/trivial_runner.py#L216-L252","documentation":"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.","triggerScenarios":"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.","commonSituations":"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.","solutions":["Avoid timers/stateful transforms when using TrivialRunner; use DirectRunner instead","Upgrade apache_beam so trivial_runner handles the new element types","Restructure the pipeline to emit only regular data outputs under the trivial runner","File/patch the runner to handle the specific Elements oneof variant"],"exampleFix":"// before\npipeline.run(runner=TrivialRunner())  # pipeline uses timers\n// after\npipeline.run(runner=DirectRunner())  # timers supported","handlingStrategy":"try-catch","validationCode":"def uses_timers_or_state(pipeline):\n    for t in pipeline.proto.components.transforms.values():\n        if 'timer' in (t.spec.urn or '').lower() or 'state' in (t.spec.urn or '').lower():\n            return True\n    return False\n# if True, don't use TrivialRunner","typeGuard":null,"tryCatchPattern":"try:\n    pipeline.run(runner=TrivialRunner()).wait_until_finish()\nexcept RuntimeError as e:\n    if 'Unexpected data type' in str(e):\n        logging.warning('TrivialRunner cannot handle this element type; using DirectRunner')\n        pipeline.run(runner=DirectRunner())\n    else:\n        raise","preventionTips":["Avoid timers/stateful DoFns when testing with TrivialRunner","Use DirectRunner for pipelines with timers or control-plane features","Keep SDK versions aligned so element payload types are understood","Check pipeline URNs for timer/state usage before picking a runner"],"tags":["apache-beam","python","runner","data-type"],"backgroundTag":"unexpected-response-shape","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"}