{"record":{"id":"892a4379b4bd913c","repo":"apache/beam","slug":"pubsub-i-o-is-only-available-in-streaming-mode-use-the","errorCode":null,"errorMessage":"PubSub I/O is only available in streaming mode (use the --streaming flag).","messagePattern":"PubSub I/O is only available in streaming mode \\(use the --streaming flag\\)\\.","errorType":"exception","errorClass":"Exception","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/direct/direct_runner.py","lineNumber":536,"sourceCode":"\n  def expand(self, pvalue):\n    # This is handled as a native transform.\n    return PCollection(self.pipeline, is_bounded=self._source.is_bounded())\n\n\ndef _get_pubsub_transform_overrides(pipeline_options):\n  from apache_beam.io.gcp import pubsub as beam_pubsub\n  from apache_beam.pipeline import PTransformOverride\n\n  class ReadFromPubSubOverride(PTransformOverride):\n    def matches(self, applied_ptransform):\n      return isinstance(\n          applied_ptransform.transform, beam_pubsub.ReadFromPubSub)\n\n    def get_replacement_transform_for_applied_ptransform(\n        self, applied_ptransform):\n      if not pipeline_options.view_as(StandardOptions).streaming:\n        raise Exception(\n            'PubSub I/O is only available in streaming mode '\n            '(use the --streaming flag).')\n      return _DirectReadFromPubSub(applied_ptransform.transform._source)\n\n  # WriteToPubSub no longer needs an override - it works by default for both\n  # batch and streaming\n  return [ReadFromPubSubOverride()]\n\n\nclass BundleBasedDirectRunner(PipelineRunner):\n  \"\"\"Executes a single pipeline on the local machine.\"\"\"\n  @staticmethod\n  def is_fnapi_compatible():\n    return False\n\n  def run_pipeline(self, pipeline, options):\n    \"\"\"Execute the entire pipeline and returns an DirectPipelineResult.\"\"\"\n","sourceCodeStart":518,"sourceCodeEnd":554,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/direct/direct_runner.py#L518-L554","documentation":"The DirectRunner's PubSub override only works in streaming mode: ReadFromPubSub requires an unbounded source and streaming execution, so in batch mode the runner raises Exception during transform replacement. The message tells you the exact flag needed.","triggerScenarios":"Running a pipeline with the DirectRunner that reads from PubSub via beam.io.ReadFromPubSub while StandardOptions.streaming is not set (no --streaming flag).","commonSituations":"Developing a streaming pipeline locally and forgetting --streaming; batch test harness accidentally including a PubSub read.","solutions":["Add --streaming (StandardOptions.view_as(StandardOptions).streaming = True) when running PubSub pipelines on the DirectRunner.","Switch to a runner that supports the desired mode, e.g. DataflowRunner with --streaming.","Remove/replace the PubSub read in batch-only pipelines with a bounded source (e.g. Create or textio)."],"exampleFix":"// before\np = beam.Pipeline(runner='DirectRunner')  # batch\n_ = p | beam.io.ReadFromPubSub(topic='projects/p/topics/t')\n// after\noptions = PipelineOptions(['--streaming'])\np = beam.Pipeline(runner='DirectRunner', options=options)\n_ = p | beam.io.ReadFromPubSub(topic='projects/p/topics/t')","handlingStrategy":"validation","validationCode":"from apache_beam.options.pipeline_options import StandardOptions\nopts = PipelineOptions(argv)\nif uses_pubsub_io(pipeline) and not opts.view_as(StandardOptions).streaming:\n    raise ValueError('PubSub reads require --streaming')","typeGuard":null,"tryCatchPattern":"try:\n    result = pipeline.run()\nexcept Exception as e:\n    if 'streaming mode' in str(e):\n        options.view_as(StandardOptions).streaming = True  # rerun with --streaming\n    else:\n        raise","preventionTips":["Always pass --streaming for pipelines with unbounded sources in local tests.","Keep batch and streaming pipeline entry points separate.","Flag pipelines containing ReadFromPubSub as streaming-only in your tooling."],"tags":["python","apache-beam","direct-runner","pubsub","streaming"],"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"}