apache/beam · error · NotImplementedError

id_label is not supported for PubSub writes with DirectRunne

Error message

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

What it means

When a Pub/Sub write runs on a runner/mode that does not support output labels (DirectRunner, or any batch execution), WriteToPubSub's expand() raises NotImplementedError if id_label was set, because record-id deduplication requires streaming support.

Source

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

        pipeline_options, 'runner',
        'None') if pipeline_options else 'No options'
    streaming_info = 'Unknown'
    if pipeline_options:
      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(

View on GitHub (pinned to 12126d8942)

Solutions

1. Remove the id_label argument when running on DirectRunner or in batch mode
2. Run on a streaming-capable runner (e.g. Dataflow with streaming=True) if dedup ids are required
3. Parameterize id_label by runner/mode so it is only set for supported executions

Example fix

// before
beam.io.WriteToPubSub(topic=t, id_label='id')  # runs on DirectRunner
// after
beam.io.WriteToPubSub(topic=t, id_label=('id' if options.view_as(SetupOptions).streaming else None))

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

try:
    pcoll | beam.io.WriteToPubSub(topic=t, id_label=id_label)
except NotImplementedError as e:
    if 'id_label is not supported' in str(e):
        pcoll | beam.io.WriteToPubSub(topic=t)  # drop id_label for this runner
    else:
        raise

When it happens

Trigger: Running a pipeline with WriteToPubSub(id_label='id', ...) on DirectRunner; running with a streaming-capable runner but in batch mode; forgetting to remove id_label when switching from Dataflow streaming to local/batch testing.

Common situations: Developers testing locally with DirectRunner after copying a production streaming pipeline definition; CI batch test runs of streaming pipelines.

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

• timestamp_attribute is not supported for PubSub writes with
• 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/e3988220d3cc0771. Report an issue: GitHub.