apache/beam · error · NotImplementedError

timestamp_attribute is not supported for PubSub writes with

Error message

timestamp_attribute is not supported for PubSub writes with DirectRunner or in batch mode (runner={runner_info}, {streaming_info})

What it means

WriteToPubSub raises NotImplementedError when timestamp_attribute is set for writes on DirectRunner or in batch mode, because attaching message timestamps from attributes requires streaming execution support.

Source

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

      try:
        standard_options = pipeline_options.view_as(StandardOptions)
        streaming_info = 'streaming=%s' % standard_options.streaming
      except Exception:
        streaming_info = 'streaming=unknown'

    logging.debug(
        'PubSub unsupported feature check: runner=%s, %s',
        runner_info,
        streaming_info)

    if not output_labels_supported:

      if transform.id_label:
        raise NotImplementedError(
            f'id_label is not supported for PubSub writes with DirectRunner '
            f'or in batch mode (runner={runner_info}, {streaming_info})')
      if transform.timestamp_attribute:
        raise NotImplementedError(
            f'timestamp_attribute is not supported for PubSub writes with '
            f'DirectRunner or in batch mode '
            f'(runner={runner_info}, {streaming_info})')

  def setup(self):
    from google.cloud import pubsub
    if self.with_ordering:
      self._pub_client = pubsub.PublisherClient(
          publisher_options=pubsub.types.PublisherOptions(
              enable_message_ordering=True,
          ))
    else:
      self._pub_client = pubsub.PublisherClient()
    self._topic = self._pub_client.topic_path(
        self.project, self.short_topic_name)

  def start_bundle(self):
    self._buffer = []

View on GitHub (pinned to 12126d8942)

Solutions

1. Remove timestamp_attribute for DirectRunner/batch executions
2. Switch to a streaming-capable runner with streaming enabled if timestamp propagation is required
3. Conditionally pass timestamp_attribute based on runner mode

Example fix

// before
beam.io.WriteToPubSub(topic=t, timestamp_attribute='ts')  # batch job
// after
beam.io.WriteToPubSub(topic=t)  # or stream on Dataflow with streaming=True

if timestamp_attribute and (runner is DirectRunner or not streaming):
    raise ValueError('timestamp_attribute unsupported on DirectRunner/batch')

try:
    pcoll | beam.io.WriteToPubSub(topic=t, timestamp_attribute=ts_attr)
except NotImplementedError as e:
    if 'timestamp_attribute is not supported' in str(e):
        pcoll | beam.io.WriteToPubSub(topic=t)
    else:
        raise

When it happens

Trigger: WriteToPubSub(timestamp_attribute='ts', ...) executed on DirectRunner; a batch Dataflow job; streaming pipelines tested in batch where the attribute was configured.

Common situations: Local testing of streaming pipelines with DirectRunner; copying streaming write transforms into batch jobs; parameterized pipelines where the runner mode changed but attributes stayed.

Understand the failure class

Background: UnsupportedOperationException and "is not supported" errors: when a library deliberately refuses a call — this error's family across 30 libraries.

Related errors

• id_label is not supported for PubSub writes with DirectRunne
• Either data (%r) or attributes (%r) must be set.
• A pubsub message data field must not exceed 10MB
• A pubsub message must not have more than 100 attributes.
• A pubsub message attribute key must not exceed 256 bytes.

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