{"record":{"id":"1db48955e4090a28","repo":"apache/beam","slug":"unexpected-output-element-type-s","errorCode":null,"errorMessage":"Unexpected output element type %s","messagePattern":"Unexpected output element type (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/data_plane.py","lineNumber":711,"sourceCode":"               len(streams) <= 100):\n          data_or_timer = self._to_send.get_nowait()\n          total_size_bytes += data_or_timer.ByteSize()\n          streams.append(data_or_timer)\n      except queue.Empty:\n        pass\n      if streams[-1] is self._WRITES_FINISHED:\n        stream_done = True\n        streams.pop()\n      if streams:\n        data_stream = []\n        timer_stream = []\n        for stream in streams:\n          if isinstance(stream, beam_fn_api_pb2.Elements.Timers):\n            timer_stream.append(stream)\n          elif isinstance(stream, beam_fn_api_pb2.Elements.Data):\n            data_stream.append(stream)\n          else:\n            raise ValueError('Unexpected output element type %s' % type(stream))\n        yield beam_fn_api_pb2.Elements(data=data_stream, timers=timer_stream)\n\n  def _get_element_size_bytes(self, element):\n    # type: (Union[beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers]) -> int\n    if isinstance(element, beam_fn_api_pb2.Elements.Data):\n      return len(element.data)\n    elif isinstance(element, beam_fn_api_pb2.Elements.Timers):\n      return len(element.timers)\n    else:\n      return 0\n\n  def _read_inputs(self, elements_iterator):\n    # type: (Iterable[beam_fn_api_pb2.Elements]) -> None\n\n    next_discard_log_time = 0  # type: float\n\n    def _put_queue(instruction_id, element):\n      # type: (str, Union[beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers]) -> None","sourceCodeStart":693,"sourceCodeEnd":729,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/data_plane.py#L693-L729","documentation":" In _write_outputs, outgoing streams are partitioned into Elements.Data and Elements.Timers protobuf sub-messages. A stream object of any other type triggers ValueError 'Unexpected output element type', protecting the data plane from writing malformed output records.","triggerScenarios":" A transform's output writer produces a stream object that is neither beam_fn_api_pb2.Elements.Data nor Elements.Timers when the data plane flushes outputs.","commonSituations":" Custom sinks/writers or patched Beam internals returning wrong stream types; SDK/runner version mismatch producing incompatible protobuf wrappers; buggy subclassing of data plane output logic.","solutions":["Verify all writer/transform code emits proper Elements.Data or Elements.Timers messages.","Align runner and SDK worker Beam versions to avoid protobuf incompatibilities.","Remove or fix any monkey-patching of data plane internals in the worker.","Reproduce with a minimal pipeline to identify which transform produces the offending output stream."],"exampleFix":"# before\nreturn SomeCustomWrapper(data)\n\n# after\nreturn beam_fn_api_pb2.Elements.Data(transform_id=transform_id, data=data)","handlingStrategy":"type-guard","validationCode":null,"typeGuard":"def is_valid_output_stream(s) -> bool:\n    import apache_beam.runners.worker.data_plane as dp\n    from apache_beam.portability.api import beam_fn_api_pb2\n    return isinstance(s, (beam_fn_api_pb2.Elements.Data, beam_fn_api_pb2.Elements.Timers))","tryCatchPattern":null,"preventionTips":["Only emit Elements.Data/Elements.Timers from custom writers.","Avoid monkey-patching data plane internals.","Keep runner and SDK protobuf definitions in sync."],"tags":["apache-beam","protobuf","type-mismatch"],"backgroundTag":"type-mismatch","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"}