{"record":{"id":"86818e6720d8214b","repo":"apache/beam","slug":"dataflow-pipeline-failed-state-s-error-s","errorCode":null,"errorMessage":"Dataflow pipeline failed. State: %s, Error:\n%s","messagePattern":"Dataflow pipeline failed\\. State: (.+?), Error:\n(.+?)","errorType":"exception","errorClass":"DataflowRuntimeException","httpStatus":null,"severity":"error","filePath":"sdks/python/apache_beam/runners/dataflow/dataflow_runner.py","lineNumber":834,"sourceCode":"      # thread will terminate everything. This is also the reason we will not\n      # use thread.join() to wait for the polling thread.\n      thread.daemon = True\n      thread.start()\n      while thread.is_alive():\n        time.sleep(5.0)\n\n      # TODO: Merge the termination code in poll_for_job_completion and\n      # is_in_terminal_state.\n      terminated = self.is_in_terminal_state()\n      assert duration or terminated, (\n          'Job did not reach to a terminal state after waiting indefinitely. '\n          '{}'.format(consoleUrl))\n\n      if terminated and self.state != PipelineState.DONE:\n        # TODO(BEAM-1290): Consider converting this to an error log based on\n        # theresolution of the issue.\n        _LOGGER.error(consoleUrl)\n        raise DataflowRuntimeException(\n            'Dataflow pipeline failed. State: %s, Error:\\n%s' %\n            (self.state, getattr(self._runner, 'last_error_msg', None)),\n            self)\n    elif PipelineState.is_terminal(\n        self.state) and self.state == PipelineState.FAILED and self._runner:\n      raise DataflowRuntimeException(\n          'Dataflow pipeline failed. State: %s, Error:\\n%s' %\n          (self.state, getattr(self._runner, 'last_error_msg', None)),\n          self)\n\n    return self.state\n\n  def cancel(self):\n    if not self.has_job:\n      raise IOError('Failed to get the Dataflow job id.')\n\n    self._update_job()\n","sourceCodeStart":816,"sourceCodeEnd":852,"githubUrl":"https://github.com/apache/beam/blob/12126d8942aaf848030c478b4c6a28c6af861c66/sdks/python/apache_beam/runners/dataflow/dataflow_runner.py#L816-L852","documentation":"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.","triggerScenarios":"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).","commonSituations":"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.","solutions":["Read the Dataflow job logs in the GCP Console (link in the logged consoleUrl) to find the root-cause worker exception","Fix the underlying pipeline error (e.g. missing package, failing user code) and resubmit","Check job state via pipeline_result.state to distinguish FAILED vs CANCELLED/DRAINED/UPDATED before treating it as a crash","If the state is CANCELLED/DRAINED intentionally, catch DataflowRuntimeException instead of letting it propagate"],"exampleFix":"# before\nresult = pipeline.run()\nresult.wait_until_finish()  # raises if state != DONE\n# after\nresult = pipeline.run()\ntry:\n    result.wait_until_finish()\nexcept DataflowRuntimeException as e:\n    print('Pipeline ended in state', result.state, '-', e)","handlingStrategy":"try-catch","validationCode":"if result.state not in (None, 'RUNNING', 'DONE'):\n    print('warning: job already in state', result.state)","typeGuard":null,"tryCatchPattern":"try:\n    state = result.wait_until_finish()\nexcept DataflowRuntimeException as e:\n    log.error('Dataflow job failed: %s (state=%s)', e, result.state)","preventionTips":["Monitor the Dataflow job in the GCP Console early to catch worker errors fast","Set up Cloud Logging alerts on job state changes to FAILED","Validate user code and dependencies with DirectRunner/FlinkRunner before submitting to Dataflow","Use pipeline_option transforms logging to surface worker exceptions in your own logs"],"tags":["dataflow","gcp","pipeline","remote-execution"],"backgroundTag":"invalid-state-transition","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"}