MaksYermak commented on code in PR #72809:
URL: https://github.com/apache/airflow/pull/72809#discussion_r4456458512
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -339,6 +339,15 @@ def _get_current_jobs(self) -> list[dict]:
jobs = self._fetch_jobs_by_prefix_name(self._job_name.lower())
if len(jobs) == 1:
self._job_id = jobs[0]["id"]
+ elif len(jobs) > 1 and not self._multiple_jobs:
+ active_jobs = [
+ job for job in jobs if job.get("currentState") not in
DataflowJobStatus.TERMINAL_STATES
+ ]
+ if len(active_jobs) == 1:
+ self._job_id = active_jobs[0]["id"]
+ else:
+ jobs.sort(key=lambda j: j.get("createTime", ""),
reverse=True)
+ self._job_id = jobs[0]["id"]
Review Comment:
@potiuk I have removed the code.
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -434,13 +443,40 @@ def fetch_job_autoscaling_events_by_id(self, job_id: str)
-> list[dict]:
autoscaling_events.extend(response.get("autoscalingEvents", []))
return autoscaling_events
- def _fetch_all_jobs(self) -> list[dict]:
- request = (
- self._dataflow.projects()
- .locations()
- .jobs()
- .list(projectId=self._project_number, location=self._job_location)
- )
+ def fetch_job_id(self) -> str | None:
+ if self._job_id:
+ return self._job_id
+ if not self._job_name:
+ raise ValueError("Missing both dataflow job ID and name.")
+ jobs = self._fetch_all_jobs(jobs_filter="ACTIVE")
+ job_name = self._job_name.lower()
+ active_jobs = [
+ job
+ for job in jobs
+ if job.get("name") == job_name
+ and job.get("currentState") not in
DataflowJobStatus.TERMINAL_STATES
+ ]
Review Comment:
> The new launcher, meanwhile, fails with DataflowJobAlreadyExistsError, and
since we've stopped reading the subprocess nobody sees it; before this PR the
task would fail.
I think this statement is incorrect. I have tested this code with parallel
run when `append_job_name=False` and it raise an
`DataflowJobAlreadyExistsError` exception and fail the Task run. Here is the
screenshot of the run:
<img width="1829" height="933" alt="Screenshot 2026-11-03 at 15 27 04"
src="https://github.com/user-attachments/assets/c02487ba-b269-450c-82d7-f93fa2103233";
/>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
