{"record":{"id":"96ad4be504cc6386","repo":"apache/beam","slug":"directrunner-id-label-is-not-supported-for-pubsub-reads","errorCode":null,"errorMessage":"DirectRunner: id_label is not supported for PubSub reads","messagePattern":"DirectRunner: id_label is not supported for PubSub reads","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/transform_evaluator.py","lineNumber":637,"sourceCode":"  _subscriber_client_cache = weakref.WeakKeyDictionary()\n  _subscriber_client_cache_lock = threading.Lock()\n\n  def __init__(\n      self,\n      evaluation_context,\n      applied_ptransform,\n      input_committed_bundle,\n      side_inputs):\n    assert not side_inputs\n    super().__init__(\n        evaluation_context,\n        applied_ptransform,\n        input_committed_bundle,\n        side_inputs)\n\n    self.source: _PubSubSource = self._applied_ptransform.transform._source\n    if self.source.id_label:\n      raise NotImplementedError(\n          'DirectRunner: id_label is not supported for PubSub reads')\n\n    sub_project = None\n    if hasattr(self._evaluation_context, 'pipeline_options'):\n      from apache_beam.options.pipeline_options import GoogleCloudOptions\n      sub_project = (\n          self._evaluation_context.pipeline_options.view_as(\n              GoogleCloudOptions).project)\n    if not sub_project:\n      sub_project = self.source.project\n\n    self._sub_name = self.get_subscription(\n        self._applied_ptransform,\n        self.source.project,\n        self.source.topic_name,\n        sub_project,\n        self.source.subscription_name)\n","sourceCodeStart":619,"sourceCodeEnd":655,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/transform_evaluator.py#L619-L655","documentation":"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.","triggerScenarios":"Creating a beam.io.ReadFromPubSub(..., id_label='some_label') (or PubSubSource with id_label set) and running the pipeline with DirectRunner.","commonSituations":"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.","solutions":["Remove the id_label argument when running under DirectRunner.","Gate the id_label option on the runner: set it only when using Dataflow.","Implement app-level deduplication (e.g. with state) instead of relying on id_label.","Run the affected test/pipeline on Dataflow or a runner supporting id_label."],"exampleFix":"# before\n_ = pipeline | ReadFromPubSub(topic='t', id_label='msg_id')\n# after\n_ = pipeline | ReadFromPubSub(topic='t')  # id_label unsupported on DirectRunner","handlingStrategy":"validation","validationCode":"if id_label and runner_type == 'DirectRunner':\n    raise ValueError('id_label is not supported on DirectRunner; drop it or use Dataflow')","typeGuard":"def pubsub_read_supported_on_direct(id_label) -> bool:\n    return not id_label","tryCatchPattern":"try:\n    pipeline.run()\nexcept NotImplementedError as e:\n    if 'id_label' in str(e):\n        rebuild_pipeline_without_id_label()","preventionTips":["Only set id_label when running on Dataflow.","Conditionally build the source based on pipeline_options.runner.","Replace id_label dedup with application-level dedup for portability."],"tags":["python","apache-beam","directrunner","pubsub"],"backgroundTag":"operation-not-supported","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-20T03:17:13.778Z"}