{"record":{"id":"39accaf5e3175488","repo":"apache/beam","slug":"unexpected-element-type-s-expected-pubsubmessage-value-r","errorCode":null,"errorMessage":"Unexpected element. Type: %s (expected: PubsubMessage), value: %r","messagePattern":"Unexpected element\\. Type: (.+?) \\(expected: PubsubMessage\\), value: %r","errorType":"validation","errorClass":"TypeError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/pubsub.py","lineNumber":426,"sourceCode":"        message with the given name and the message's publish time as the value.\n      publish_with_ordering_key: If True, enables message ordering on the\n        PublisherClient. Messages with an ordering_key will be delivered\n        in order. Requires messages to have ordering_key set.\n    \"\"\"\n    super().__init__()\n    self.with_attributes = with_attributes\n    self.publish_with_ordering_key = publish_with_ordering_key\n    self.id_label = id_label\n    self.timestamp_attribute = timestamp_attribute\n    self.project, self.topic_name = parse_topic(topic)\n    self.full_topic = topic\n    self._sink = _PubSubSink(topic, id_label, timestamp_attribute)\n    self.pipeline_options = None  # Will be set during expand()\n\n  @staticmethod\n  def message_to_proto_str(element: PubsubMessage) -> bytes:\n    if not isinstance(element, PubsubMessage):\n      raise TypeError(\n          'Unexpected element. Type: %s (expected: PubsubMessage), '\n          'value: %r' % (type(element), element))\n    return element._to_proto_str(for_publish=True)\n\n  @staticmethod\n  def bytes_to_proto_str(element: Union[bytes, str]) -> bytes:\n    msg = PubsubMessage(element, {})\n    return msg._to_proto_str(for_publish=True)\n\n  def expand(self, pcoll):\n    # Store pipeline options for use in DoFn\n    self.pipeline_options = pcoll.pipeline.options if pcoll.pipeline else None\n    # Warn Dataflow users to use the XLang path for ordering key support,\n    # since _PubSubWriteDoFn._flush() is not used by Dataflow's implementation.\n    runner = self.pipeline_options.get_all_options().get(\n        'runner', '') if self.pipeline_options else ''\n    if 'Dataflow' in str(runner) and self.publish_with_ordering_key:\n      logging.warning(","sourceCodeStart":408,"sourceCodeEnd":444,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/pubsub.py#L408-L444","documentation":"WriteToPubSub's message_to_proto_str static method only accepts apache_beam.io.gcp.pubsub.PubsubMessage elements. When a PTransform pipeline element handed to PubSub write is not a PubsubMessage instance (e.g. raw bytes, str, or dict), a TypeError is raised with the offending type and value repr.","triggerScenarios":"Passing a bytes/str/dict element into WriteToPubSub with to_callable defaulting to message_to_proto_str; mixing pipelines where the writer is used on non-PubsubMessage PCollections; upgrading Beam versions where implicit conversion of bytes/str was removed.","commonSituations":"Developers migrating old code that published raw byte strings to Pub/Sub via WriteToPubSub; constructing PubsubMessage manually but passing the wrong wrapper type; testing pipelines with dummy string data.","solutions":["Wrap the element in PubsubMessage(data, attributes) before writing","Pass an explicit to_callable that converts your element type to serialized proto bytes, e.g. to_callable=lambda e: e.SerializeToString() or bytes-based encoding","Check the pipeline output type: ensure the producing PCollection actually yields PubsubMessage objects"],"exampleFix":"// before\npcoll | 'write' beam.io.WriteToPubSub(topic='projects/p/topics/t')  # pcoll of bytes\n// after\npcoll | beam.Map(lambda b: PubsubMessage(b, {})) | 'write' beam.io.WriteToPubSub(topic='projects/p/topics/t')","handlingStrategy":"type-guard","validationCode":"from apache_beam.io.gcp.pubsub import PubsubMessage\nassert isinstance(el, PubsubMessage), f'expected PubsubMessage, got {type(el)}'","typeGuard":"def is_pubsub_message(el) -> bool:\n    from apache_beam.io.gcp.pubsub import PubsubMessage\n    return isinstance(el, PubsubMessage)","tryCatchPattern":"try:\n    result = pcoll | beam.io.WriteToPubSub(topic=t)\nexcept TypeError as e:\n    if 'expected: PubsubMessage' in str(e):\n        pcoll = pcoll | beam.Map(lambda b: PubsubMessage(b, {})) | beam.io.WriteToPubSub(topic=t)\n    else:\n        raise","preventionTips":["Always wrap raw payloads in PubsubMessage before writing to Pub/Sub","Pin a single to_callable for the write transform and document the expected element type","Add a beam.Map(isinstance-check) assert stage during local testing"],"tags":["python","pubsub","type-mismatch","apache-beam"],"backgroundTag":"type-mismatch","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}