Skip to content

Dataflow job is failed when wait_until_finished=True although the state is JOB_STATE_DONE #34767

Description

@champon1020

Apache Airflow version

2.7.1

What happened

We currently use the DataflowHook in tasks of Airflow DAG. If we upgrade the version of apache-airflow-google-providers to 10.9.0, we got the following error although the dataflow job is completed.

Traceback (most recent call last):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1384, in _run_raw_task
    self._execute_task_with_callbacks(context, test_mode)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1531, in _execute_task_with_callbacks
    result = self._execute_task(context, task_orig)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1586, in _execute_task
    result = execute_callable(context=context)
  File "xxx", line 65, in execute
    hook.wait_for_done(
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/common/hooks/base_google.py", line 475, in inner_wrapper
    return func(self, *args, **kwargs)
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 1203, in wait_for_done
    job_controller.wait_for_done()
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 439, in wait_for_done
    while self._jobs and not all(self._check_dataflow_job_state(job) for job in self._jobs):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 439, in <genexpr>
    while self._jobs and not all(self._check_dataflow_job_state(job) for job in self._jobs):
  File "/opt/python3.8/lib/python3.8/site-packages/airflow/providers/google/cloud/hooks/dataflow.py", line 430, in _check_dataflow_job_state
    raise Exception(
Exception: Google Cloud Dataflow job <xxx> is in an unexpected terminal state: JOB_STATE_DONE, expected terminal state: JOB_STATE_DONE

What you think should happen instead

The error message "an unexpected terminal state: JOB_STATE_DONE, expected terminal state: JOB_STATE_DONE" is strange. If the dataflow job is completed, I think it should not be failed even if the expected_terminal_state is not set as DataflowHook parameter.

How to reproduce

Install airflow from apache-airflow-google-providers/10.9.0.
Pass wait_until_finished=True to DataflowHook and execute start_template_dataflow.

Operating System

Ubuntu 20.04.6 LTS (Focal Fossa)

Versions of Apache Airflow Providers

apache-airflow-google-providers===10.9.0

Deployment

Google Cloud Composer

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

  • Yes I am willing to submit a PR!

Code of Conduct

Metadata

Metadata

Assignees

No one assigned

    Labels

    Type

    No type

    Projects

    No projects

    Milestone

    No milestone

    Relationships

    None yet

    Development

    No branches or pull requests

    Issue actions