{"record":{"id":"b5ad0f4552bff0df","repo":"apache/beam","slug":"timestamp-attribute-is-not-supported-for-pubsub-writes-with","errorCode":null,"errorMessage":"timestamp_attribute is not supported for PubSub writes with DirectRunner or in batch mode (runner={runner_info}, {streaming_info})","messagePattern":"timestamp_attribute 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":643,"sourceCode":"      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(\n        self.project, self.short_topic_name)\n\n  def start_bundle(self):\n    self._buffer = []","sourceCodeStart":625,"sourceCodeEnd":661,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/io/gcp/pubsub.py#L625-L661","documentation":"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.","triggerScenarios":"WriteToPubSub(timestamp_attribute='ts', ...) executed on DirectRunner; a batch Dataflow job; streaming pipelines tested in batch where the attribute was configured.","commonSituations":"Local testing of streaming pipelines with DirectRunner; copying streaming write transforms into batch jobs; parameterized pipelines where the runner mode changed but attributes stayed.","solutions":["Remove timestamp_attribute for DirectRunner/batch executions","Switch to a streaming-capable runner with streaming enabled if timestamp propagation is required","Conditionally pass timestamp_attribute based on runner mode"],"exampleFix":"// before\nbeam.io.WriteToPubSub(topic=t, timestamp_attribute='ts')  # batch job\n// after\nbeam.io.WriteToPubSub(topic=t)  # or stream on Dataflow with streaming=True","handlingStrategy":"validation","validationCode":"if timestamp_attribute and (runner is DirectRunner or not streaming):\n    raise ValueError('timestamp_attribute unsupported on DirectRunner/batch')","typeGuard":null,"tryCatchPattern":"try:\n    pcoll | beam.io.WriteToPubSub(topic=t, timestamp_attribute=ts_attr)\nexcept NotImplementedError as e:\n    if 'timestamp_attribute is not supported' in str(e):\n        pcoll | beam.io.WriteToPubSub(topic=t)\n    else:\n        raise","preventionTips":["Conditionally pass timestamp_attribute only for streaming executions","Add a runner/mode smoke test to CI that exercises write transforms","Keep streaming-specific args out of shared transform factories"],"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"}