apache/beam · error · NotImplementedError

DirectRunner: id_label is not supported for PubSub reads

Error message

DirectRunner: id_label is not supported for PubSub reads

What it means

The DirectRunner Pub/Sub read evaluator does not support exactly-once id_label-based deduplication. If the PubSub source was configured with id_label, __init__ raises NotImplementedError since local reads can't emulate that service feature.

Solutions

1. Remove the id_label argument when running under DirectRunner.
2. Gate the id_label option on the runner: set it only when using Dataflow.
3. Implement app-level deduplication (e.g. with state) instead of relying on id_label.
4. Run the affected test/pipeline on Dataflow or a runner supporting id_label.

Example fix

# before
_ = pipeline | ReadFromPubSub(topic='t', id_label='msg_id')
# after
_ = pipeline | ReadFromPubSub(topic='t')  # id_label unsupported on DirectRunner

if id_label and runner_type == 'DirectRunner':
    raise ValueError('id_label is not supported on DirectRunner; drop it or use Dataflow')

def pubsub_read_supported_on_direct(id_label) -> bool:
    return not id_label

try:
    pipeline.run()
except NotImplementedError as e:
    if 'id_label' in str(e):
        rebuild_pipeline_without_id_label()

When it happens

Trigger: Creating a beam.io.ReadFromPubSub(..., id_label='some_label') (or PubSubSource with id_label set) and running the pipeline with DirectRunner.

Common situations: Pipelines authored for Dataflow (where id_label dedup works) run locally in tests; copying example code that sets id_label; migrating a pipeline from Dataflow to local execution.

Related errors

• A pubsub message attribute key must not exceed 256 bytes.
• A pubsub message attribute value must not exceed 1024 bytes
• A pubsub message data field must not exceed 10MB
• A pubsub message must not have more than 100 attributes.
• A pubsub message ordering key must not exceed 1024 bytes.

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/direct/transform_evaluator.py:637

  _subscriber_client_cache = weakref.WeakKeyDictionary()
  _subscriber_client_cache_lock = threading.Lock()

  def __init__(
      self,
      evaluation_context,
      applied_ptransform,
      input_committed_bundle,
      side_inputs):
    assert not side_inputs
    super().__init__(
        evaluation_context,
        applied_ptransform,
        input_committed_bundle,
        side_inputs)

    self.source: _PubSubSource = self._applied_ptransform.transform._source
    if self.source.id_label:
      raise NotImplementedError(
          'DirectRunner: id_label is not supported for PubSub reads')

    sub_project = None
    if hasattr(self._evaluation_context, 'pipeline_options'):
      from apache_beam.options.pipeline_options import GoogleCloudOptions
      sub_project = (
          self._evaluation_context.pipeline_options.view_as(
              GoogleCloudOptions).project)
    if not sub_project:
      sub_project = self.source.project

    self._sub_name = self.get_subscription(
        self._applied_ptransform,
        self.source.project,
        self.source.topic_name,
        sub_project,
        self.source.subscription_name)

View on GitHub (pinned to 12126d8942)