This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8442ef02563 Remove unused Dataflow job state checker (#75784)
8442ef02563 is described below
commit 8442ef025631de285ca1ff570c1eb33ff2ec9382
Author: Hussein Awala <[email protected]>
AuthorDate: Tue Oct 27 21:54:33 2026 +0100
Remove unused Dataflow job state checker (#75784)
The private controller polls through inherited job_reached_terminal_state;
the deleted checker has no callers.
---
generated/known_airflow_exceptions.txt | 2 +-
.../providers/google/cloud/hooks/dataflow.py | 49 ----------------------
2 files changed, 1 insertion(+), 50 deletions(-)
diff --git a/generated/known_airflow_exceptions.txt
b/generated/known_airflow_exceptions.txt
index 47ebd560a73..41bafef7fc3 100644
--- a/generated/known_airflow_exceptions.txt
+++ b/generated/known_airflow_exceptions.txt
@@ -213,7 +213,7 @@
providers/google/src/airflow/providers/google/cloud/hooks/cloud_sql.py::31
providers/google/src/airflow/providers/google/cloud/hooks/cloud_storage_transfer_service.py::5
providers/google/src/airflow/providers/google/cloud/hooks/compute.py::6
providers/google/src/airflow/providers/google/cloud/hooks/compute_ssh.py::6
-providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py::13
+providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py::9
providers/google/src/airflow/providers/google/cloud/hooks/dataform.py::4
providers/google/src/airflow/providers/google/cloud/hooks/dataplex.py::2
providers/google/src/airflow/providers/google/cloud/hooks/dataproc.py::5
diff --git
a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
index ea1af01c1ad..e3fd6da93fc 100644
--- a/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
+++ b/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py
@@ -480,55 +480,6 @@ class
_DataflowJobsController(DataflowJobTerminalStateHelper):
else:
self.log.info("Google Cloud DataFlow job not available yet..")
- def _check_dataflow_job_state(self, job) -> bool:
- """
- Check the state of one job in dataflow for this task if job failed
raise exception.
-
- :return: True if job is done.
- :raise: Exception
- """
- current_state = job["currentState"]
- is_streaming = job.get("type") == DataflowJobType.JOB_TYPE_STREAMING
-
- current_expected_state = self._expected_terminal_state
-
- if current_expected_state is None:
- if is_streaming:
- current_expected_state = DataflowJobStatus.JOB_STATE_RUNNING
- else:
- current_expected_state = DataflowJobStatus.JOB_STATE_DONE
-
- terminal_states = DataflowJobStatus.TERMINAL_STATES |
{DataflowJobStatus.JOB_STATE_RUNNING}
- if current_expected_state not in terminal_states:
- raise AirflowException(
- f"Google Cloud Dataflow job's expected terminal state "
- f"'{current_expected_state}' is invalid."
- f" The value should be any of the following: {terminal_states}"
- )
- if is_streaming and current_expected_state ==
DataflowJobStatus.JOB_STATE_DONE:
- raise AirflowException(
- "Google Cloud Dataflow job's expected terminal state cannot be
"
- "JOB_STATE_DONE while it is a streaming job"
- )
- if not is_streaming and current_expected_state ==
DataflowJobStatus.JOB_STATE_DRAINED:
- raise AirflowException(
- "Google Cloud Dataflow job's expected terminal state cannot be
"
- "JOB_STATE_DRAINED while it is a batch job"
- )
- if current_state == current_expected_state:
- if current_expected_state == DataflowJobStatus.JOB_STATE_RUNNING:
- return not self._wait_until_finished
- return True
-
- if current_state in DataflowJobStatus.AWAITING_STATES:
- return self._wait_until_finished is False
-
- self.log.debug("Current job: %s", job)
- raise AirflowException(
- f"Google Cloud Dataflow job {job['name']} is in an unexpected
terminal state: {current_state}, "
- f"expected terminal state: {current_expected_state}"
- )
-
def wait_for_done(self) -> None:
"""Wait for result of submitted job."""
self.log.info("Start waiting for done.")
