apache/beam · error · TypeError

Unexpected element. Type: %s (expected: PubsubMessage), valu

Error message

Unexpected element. Type: %s (expected: PubsubMessage), value: %r

What it means

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.

Source

Thrown at sdks/python/apache_beam/io/gcp/pubsub.py:426

        message with the given name and the message's publish time as the value.
      publish_with_ordering_key: If True, enables message ordering on the
        PublisherClient. Messages with an ordering_key will be delivered
        in order. Requires messages to have ordering_key set.
    """
    super().__init__()
    self.with_attributes = with_attributes
    self.publish_with_ordering_key = publish_with_ordering_key
    self.id_label = id_label
    self.timestamp_attribute = timestamp_attribute
    self.project, self.topic_name = parse_topic(topic)
    self.full_topic = topic
    self._sink = _PubSubSink(topic, id_label, timestamp_attribute)
    self.pipeline_options = None  # Will be set during expand()

  @staticmethod
  def message_to_proto_str(element: PubsubMessage) -> bytes:
    if not isinstance(element, PubsubMessage):
      raise TypeError(
          'Unexpected element. Type: %s (expected: PubsubMessage), '
          'value: %r' % (type(element), element))
    return element._to_proto_str(for_publish=True)

  @staticmethod
  def bytes_to_proto_str(element: Union[bytes, str]) -> bytes:
    msg = PubsubMessage(element, {})
    return msg._to_proto_str(for_publish=True)

  def expand(self, pcoll):
    # Store pipeline options for use in DoFn
    self.pipeline_options = pcoll.pipeline.options if pcoll.pipeline else None
    # Warn Dataflow users to use the XLang path for ordering key support,
    # since _PubSubWriteDoFn._flush() is not used by Dataflow's implementation.
    runner = self.pipeline_options.get_all_options().get(
        'runner', '') if self.pipeline_options else ''
    if 'Dataflow' in str(runner) and self.publish_with_ordering_key:
      logging.warning(

View on GitHub (pinned to 12126d8942)

Solutions

1. Wrap the element in PubsubMessage(data, attributes) before writing
2. 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
3. Check the pipeline output type: ensure the producing PCollection actually yields PubsubMessage objects

Example fix

// before
pcoll | 'write' beam.io.WriteToPubSub(topic='projects/p/topics/t')  # pcoll of bytes
// after
pcoll | beam.Map(lambda b: PubsubMessage(b, {})) | 'write' beam.io.WriteToPubSub(topic='projects/p/topics/t')

from apache_beam.io.gcp.pubsub import PubsubMessage
assert isinstance(el, PubsubMessage), f'expected PubsubMessage, got {type(el)}'

def is_pubsub_message(el) -> bool:
    from apache_beam.io.gcp.pubsub import PubsubMessage
    return isinstance(el, PubsubMessage)

try:
    result = pcoll | beam.io.WriteToPubSub(topic=t)
except TypeError as e:
    if 'expected: PubsubMessage' in str(e):
        pcoll = pcoll | beam.Map(lambda b: PubsubMessage(b, {})) | beam.io.WriteToPubSub(topic=t)
    else:
        raise

When it happens

Trigger: 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.

Common situations: 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.

Understand the failure class

Background: Type mismatch errors: IllegalArgumentException, TypeError and type guards across 150 open-source libraries — this error's family across 150 libraries.

Related errors

• Cannot encode payload for WriteToPubSub. Expected valid stri
• serializable Row does not exist for payload of type: %s
• repeat(repeats=) value must be an int or a DeferredSeries (e
• Passing a deferred series to round() is not supported, pleas
• str.repeat(repeats=) value must be an int or a DeferredSerie

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