{"record":{"id":"d7cfefb2f006ddd7","repo":"apache/beam","slug":"pipeline-failed","errorCode":null,"errorMessage":"Pipeline failed.","messagePattern":"Pipeline failed\\.","errorType":"exception","errorClass":"RuntimeError","httpStatus":null,"severity":"critical","filePath":"sdks/python/apache_beam/runners/portability/portable_runner.py","lineNumber":575,"sourceCode":"    if duration:\n      state_thread = threading.Thread(\n          target=functools.partial(self._observe_state, message_thread),\n          name='wait_until_finish_state_observer')\n      state_thread.daemon = True\n      state_thread.start()\n      start_time = time.time()\n      duration_secs = duration / 1000\n      while (time.time() - start_time < duration_secs and\n             state_thread.is_alive()):\n        time.sleep(1)\n    else:\n      self._observe_state(message_thread)\n\n    if self._runtime_exception:\n      raise self._runtime_exception\n    from apache_beam.runners.runner import PipelineState\n    if self._state == PipelineState.FAILED:\n      raise RuntimeError(last_error_text or \"Pipeline failed.\")\n\n    return self._state\n\n  def _observe_state(self, message_thread):\n    try:\n      for state_response in self._state_stream:\n        self._state = self.runner_api_state_to_pipeline_state(\n            state_response.state)\n        if state_response.state in TERMINAL_STATES:\n          # Wait for any last messages.\n          message_thread.join(10)\n          break\n      if self._state != runner.PipelineState.DONE:\n        self._runtime_exception = RuntimeError(\n            'Pipeline %s failed in state %s: %s' %\n            (self._job_id, self._state, self._last_error_message()))\n    except Exception as e:\n      self._runtime_exception = e","sourceCodeStart":557,"sourceCodeEnd":593,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/portability/portable_runner.py#L557-L593","documentation":"wait_until_finish() raises this RuntimeError when the job service reports the pipeline reached the FAILED state and no more specific runtime exception was captured. It surfaces remote runner failures (worker crashes, user code exceptions) to the submitting client.","triggerScenarios":"Calling pipeline_result.wait_until_finish() (or run() waiting inline) when the remote job transitions to FAILED; last_error_text was empty so the generic 'Pipeline failed.' message is used.","commonSituations":"User code raising inside DoFns on a remote runner (Flink/Spark/prism); worker OOM or resource exhaustion; deserialization failures of closures/dependencies on the job service.","solutions":["Inspect the job service / runner logs (and any last_error_text in the message stream) for the root-cause exception.","Fix the failing pipeline code (often a user DoFn exception or missing dependency on workers).","Catch RuntimeError around wait_until_finish and use pipeline_result.state / messages for diagnostics.","Retry transient infrastructure failures (worker OOM, network) after adjusting resources."],"exampleFix":"try:\n    result.wait_until_finish()\nexcept RuntimeError:\n    print('Job failed:', result.state)\n    for msg in result.metrics_io_error_messages():\n        print(msg)","handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"result = pipeline.run()\ntry:\n    result.wait_until_finish()\nexcept RuntimeError as e:\n    print('job failed:', result.state, e)\n    raise","preventionTips":["Log job-service/runner messages (result messages) to capture the root cause early.","Ensure all pipeline dependencies are available on workers.","Right-size worker memory to avoid OOM kills.","Test pipelines locally with DirectRunner before remote submission."],"tags":["remote-runner","job-failure","pipeline"],"backgroundTag":"pipeline-execution-failed","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"}