apache/beam · error · DataflowRuntimeException

Dataflow pipeline failed. State

Error message

Dataflow pipeline failed. State: %s, Error:
%s

What it means

DataflowRuntimeException raised by DataflowPipelineResult.wait_until_finish() when the pipeline reached a terminal state other than DONE (typically FAILED). The runner surfaces the final job state and the last error message fetched from the Dataflow service. It is the canonical 'your pipeline failed on the service' error.

Solutions

  1. Read the Dataflow job logs in the GCP Console (link in the logged consoleUrl) to find the root-cause worker exception
  2. Fix the underlying pipeline error (e.g. missing package, failing user code) and resubmit
  3. Check job state via pipeline_result.state to distinguish FAILED vs CANCELLED/DRAINED/UPDATED before treating it as a crash
  4. If the state is CANCELLED/DRAINED intentionally, catch DataflowRuntimeException instead of letting it propagate

Example fix

# before
result = pipeline.run()
result.wait_until_finish()  # raises if state != DONE
# after
result = pipeline.run()
try:
    result.wait_until_finish()
except DataflowRuntimeException as e:
    print('Pipeline ended in state', result.state, '-', e)
Defensive patterns

Strategy: try-catch

Validate before calling

if result.state not in (None, 'RUNNING', 'DONE'):
    print('warning: job already in state', result.state)

Try / catch

try:
    state = result.wait_until_finish()
except DataflowRuntimeException as e:
    log.error('Dataflow job failed: %s (state=%s)', e, result.state)

Prevention

When it happens

Trigger: Calling pipeline_result.wait_until_finish() (or .result()) after submitting a job to Dataflow, and the remote job transitions to a terminal state that is not JOB_STATE_DONE (e.g. JOB_STATE_FAILED, JOB_STATE_CANCELLED, JOB_STATE_DRAINED, JOB_STATE_UPDATED).

Common situations: Worker crashes (OOM, missing dependencies), user code throwing exceptions in DoFns, quota/permission issues on GCP resources, bad side-input or sink configuration, job cancelled by another actor, or drained by an update operation.

Understand the failure class

Background: "Invalid state transition" errors: "status must be X, actually Y", "already rejected/charging/uninstalled", "cannot ... while running" — what they mean when a library rejects your call — this error's family across 31 libraries.

Related errors

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

Appendix: source

Thrown at sdks/python/apache_beam/runners/dataflow/dataflow_runner.py:834

      # thread will terminate everything. This is also the reason we will not
      # use thread.join() to wait for the polling thread.
      thread.daemon = True
      thread.start()
      while thread.is_alive():
        time.sleep(5.0)

      # TODO: Merge the termination code in poll_for_job_completion and
      # is_in_terminal_state.
      terminated = self.is_in_terminal_state()
      assert duration or terminated, (
          'Job did not reach to a terminal state after waiting indefinitely. '
          '{}'.format(consoleUrl))

      if terminated and self.state != PipelineState.DONE:
        # TODO(BEAM-1290): Consider converting this to an error log based on
        # theresolution of the issue.
        _LOGGER.error(consoleUrl)
        raise DataflowRuntimeException(
            'Dataflow pipeline failed. State: %s, Error:\n%s' %
            (self.state, getattr(self._runner, 'last_error_msg', None)),
            self)
    elif PipelineState.is_terminal(
        self.state) and self.state == PipelineState.FAILED and self._runner:
      raise DataflowRuntimeException(
          'Dataflow pipeline failed. State: %s, Error:\n%s' %
          (self.state, getattr(self._runner, 'last_error_msg', None)),
          self)

    return self.state

  def cancel(self):
    if not self.has_job:
      raise IOError('Failed to get the Dataflow job id.')

    self._update_job()

View on GitHub (pinned to 12126d8942)