{"record":{"id":"e3988220d3cc0771","repo":"apache/beam","slug":"id-label-is-not-supported-for-pubsub-writes-with","errorCode":null,"errorMessage":"id_label is not supported for PubSub writes with DirectRunner or in batch mode (runner={runner_info}, {streaming_info})","messagePattern":"id_label is not supported for PubSub writes with DirectRunner or in batch mode \\(runner=(.+?), (.+?)\\)","errorType":"exception","errorClass":"NotImplementedError","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/io/gcp/pubsub.py","lineNumber":639,"sourceCode":"        pipeline_options, 'runner',\n        'None') if pipeline_options else 'No options'\n    streaming_info = 'Unknown'\n    if pipeline_options:\n      try:\n        standard_options = pipeline_options.view_as(StandardOptions)\n        streaming_info = 'streaming=%s' % standard_options.streaming\n      except Exception:\n        streaming_info = 'streaming=unknown'\n\n    logging.debug(\n        'PubSub unsupported feature check: runner=%s, %s',\n        runner_info,\n        streaming_info)\n\n    if not output_labels_supported:\n\n      if transform.id_label:\n        raise NotImplementedError(\n            f'id_label is not supported for PubSub writes with DirectRunner '\n            f'or in batch mode (runner={runner_info}, {streaming_info})')\n      if transform.timestamp_attribute:\n        raise NotImplementedError(\n            f'timestamp_attribute is not supported for PubSub writes with '\n            f'DirectRunner or in batch mode '\n            f'(runner={runner_info}, {streaming_info})')\n\n  def setup(self):\n    from google.cloud import pubsub\n    if self.with_ordering:\n      self._pub_client = pubsub.PublisherClient(\n          publisher_options=pubsub.types.PublisherOptions(\n              enable_message_ordering=True,\n          ))\n    else:\n      self._pub_client = pubsub.PublisherClient()\n    self._topic = self._pub_client.topic_path(","sourceCodeStart":621,"sourceCodeEnd":657,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/pubsub.py#L621-L657","documentation":"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.","triggerScenarios":"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.","commonSituations":"Developers testing locally with DirectRunner after copying a production streaming pipeline definition; CI batch test runs of streaming pipelines.","solutions":["Remove the id_label argument when running on DirectRunner or in batch mode","Run on a streaming-capable runner (e.g. Dataflow with streaming=True) if dedup ids are required","Parameterize id_label by runner/mode so it is only set for supported executions"],"exampleFix":"// before\nbeam.io.WriteToPubSub(topic=t, id_label='id')  # runs on DirectRunner\n// after\nbeam.io.WriteToPubSub(topic=t, id_label=('id' if options.view_as(SetupOptions).streaming else None))","handlingStrategy":"validation","validationCode":"if id_label and (runner is DirectRunner or not streaming):\n    raise ValueError('id_label unsupported on DirectRunner/batch')","typeGuard":null,"tryCatchPattern":"try:\n    pcoll | beam.io.WriteToPubSub(topic=t, id_label=id_label)\nexcept NotImplementedError as e:\n    if 'id_label is not supported' in str(e):\n        pcoll | beam.io.WriteToPubSub(topic=t)  # drop id_label for this runner\n    else:\n        raise","preventionTips":["Gate id_label on pipeline options streaming flag","Test streaming pipelines in batch mode with attributes stripped","Pin runner assumptions next to transform construction"],"tags":["python","pubsub","unsupported-operation","apache-beam"],"backgroundTag":"unsupported-operation","analyzedSha":"12126d8942aaf848030c478b4c6a28c6af861c66","analyzedAt":"2026-09-13T01:50:10.254Z","contentChangedAt":"2026-09-13T01:50:10.254Z","schemaVersion":2},"datasetVersion":"2026-09-14T21:17:11.552Z"}