{"record":{"id":"e4f7ab7b385aaac1","repo":"apache/beam","slug":"unexpected-input-element-type-s","errorCode":null,"errorMessage":"Unexpected input element type %s","messagePattern":"Unexpected input element type (.+?)","errorType":"exception","errorClass":"ValueError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/worker/data_plane.py","lineNumber":625,"sourceCode":"                instruction_id,\n                current_time - start_time)\n            next_waiting_log_time = current_time + log_interval_sec\n        else:\n          start_time = time.time()\n          next_waiting_log_time = start_time + log_interval_sec\n          if isinstance(element, beam_fn_api_pb2.Elements.Timers):\n            if element.is_last:\n              done_inputs.add((element.transform_id, element.timer_family_id))\n            else:\n              yield element\n          elif isinstance(element, beam_fn_api_pb2.Elements.Data):\n            if element.is_last:\n              done_inputs.add(element.transform_id)\n            else:\n              assert element.transform_id not in done_inputs\n              yield element\n          else:\n            raise ValueError('Unexpected input element type %s' % type(element))\n    finally:\n      # Instruction_ids are not reusable so Clean queue once we are done with\n      #  an instruction_id\n      self._clean_receiving_queue(instruction_id)\n\n  def output_stream(self, instruction_id, transform_id):\n    # type: (str, str) -> ClosableOutputStream\n    def add_to_send_queue(data):\n      # type: (bytes) -> None\n      if data:\n        elem = beam_fn_api_pb2.Elements.Data(\n            instruction_id=instruction_id, transform_id=transform_id, data=data)\n        self._enqueue_to_send(elem)\n\n    def close_callback(data):\n      # type: (bytes) -> None\n      add_to_send_queue(data)\n      # End of stream marker.","sourceCodeStart":607,"sourceCodeEnd":643,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/worker/data_plane.py#L607-L643","documentation":" 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.","triggerScenarios":" 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).","commonSituations":" 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.","solutions":["Ensure runner and SDK harness use the same Apache Beam version (pin versions across the deployment).","Check for accidental downgrades/upgrade of the beam package at runtime in the worker image.","Inspect the payload producer (custom sources/transforms) that enqueues elements into the data plane.","File/inspect a Beam issue if reproducible with stock components; this should be unreachable in normal use."],"exampleFix":"// before\n# runner: apache-beam==2.50.0, worker image: apache-beam==2.58.0\n// after\n# pin both to the same version, e.g.\n# runner: apache-beam==2.58.0, worker image: apache-beam==2.58.0","handlingStrategy":"validation","validationCode":"# verify version parity before submitting\nimport apache_beam as beam\nassert runner_beam_version == beam.__version__, 'runner/SDK Beam version mismatch'","typeGuard":null,"tryCatchPattern":null,"preventionTips":["Pin identical apache-beam versions on runner and worker images.","Scan worker startup logs for pip downgrades of apache-beam.","Rebuild worker images on every Beam upgrade."],"tags":["apache-beam","protobuf","version-skew"],"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"}