apache/beam · critical · RuntimeError

Pipeline failed.

Error message

Pipeline failed.

What it means

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.

Solutions

1. Inspect the job service / runner logs (and any last_error_text in the message stream) for the root-cause exception.
2. Fix the failing pipeline code (often a user DoFn exception or missing dependency on workers).
3. Catch RuntimeError around wait_until_finish and use pipeline_result.state / messages for diagnostics.
4. Retry transient infrastructure failures (worker OOM, network) after adjusting resources.

Example fix

try:
    result.wait_until_finish()
except RuntimeError:
    print('Job failed:', result.state)
    for msg in result.metrics_io_error_messages():
        print(msg)

result = pipeline.run()
try:
    result.wait_until_finish()
except RuntimeError as e:
    print('job failed:', result.state, e)
    raise

When it happens

Trigger: 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.

Common situations: 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.

Related errors

• BigQuery job failed. Error Result
• Blocking computation failed. State
• Dataflow pipeline failed. State
• DLQ Router only supports PCollectionTuples split between…
• empty graph

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/d7cfefb2f006ddd7. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/runners/portability/portable_runner.py:575

    if duration:
      state_thread = threading.Thread(
          target=functools.partial(self._observe_state, message_thread),
          name='wait_until_finish_state_observer')
      state_thread.daemon = True
      state_thread.start()
      start_time = time.time()
      duration_secs = duration / 1000
      while (time.time() - start_time < duration_secs and
             state_thread.is_alive()):
        time.sleep(1)
    else:
      self._observe_state(message_thread)

    if self._runtime_exception:
      raise self._runtime_exception
    from apache_beam.runners.runner import PipelineState
    if self._state == PipelineState.FAILED:
      raise RuntimeError(last_error_text or "Pipeline failed.")

    return self._state

  def _observe_state(self, message_thread):
    try:
      for state_response in self._state_stream:
        self._state = self.runner_api_state_to_pipeline_state(
            state_response.state)
        if state_response.state in TERMINAL_STATES:
          # Wait for any last messages.
          message_thread.join(10)
          break
      if self._state != runner.PipelineState.DONE:
        self._runtime_exception = RuntimeError(
            'Pipeline %s failed in state %s: %s' %
            (self._job_id, self._state, self._last_error_message()))
    except Exception as e:
      self._runtime_exception = e

View on GitHub (pinned to 12126d8942)