aaron-y-chen commented on code in PR #72809:
URL: https://github.com/apache/airflow/pull/72809#discussion_r4023666984
##########
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:
With `append_job_name=False`, the [job name stays the
same](https://github.com/apache/airflow/blob/be95313ae7d67eb04868f381f61087df62df2ec5/providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py#L1066-L1069)
for every run.
For example, this Dag could hit the issue.
```python
BeamRunPythonPipelineOperator(
task_id="run_beam",
runner="DataflowRunner",
py_file="gs://my-bucket/etl.py",
deferrable=True,
dataflow_config=DataflowConfiguration(
job_name="daily-etl",
append_job_name=False,
location="us-central1",
),
)
```
Step 1.
The lookup runs at the first [`select`
timeout](https://github.com/apache/airflow/blob/be95313ae7d67eb04868f381f61087df62df2ec5/providers/apache/beam/src/airflow/providers/apache/beam/hooks/beam.py#L184)(5
seconds), before this run's job is created, since staging takes much longer
than that.
So `_fetch_jobs_by_prefix_name()` returns every job in the region, including
finished ones, and keeps the ones whose name starts with `daily-etl`, like the
jobs from the past two days. `jobs` would look something like this:
```python
jobs = [
{"id": "2026-09-14_03_00_12-1182736549827",
"name": "daily-etl", "currentState": "JOB_STATE_DONE",
"createTime": "2026-09-14T03:00:12Z"},
{"id": "2026-09-15_03_00_09-9938174462051",
"name": "daily-etl", "currentState": "JOB_STATE_DONE",
"createTime": "2026-09-15T03:00:09Z"}
]
```
Step 2.
Since both are in `DataflowJobStatus.TERMINAL_STATES`, `active_jobs` ends up
empty, so we fall nto the `else` branch and `self._job_id` gets yesterday's job:
```python
else:
jobs.sort(key=lambda j: j.get("createTime", ""), reverse=True)
# "2026-09-15T03:00:09Z"
# "2026-09-14T03:00:12Z"
self._job_id = jobs[0]["id"]
# = "2026-09-15_03_00_09-9938174462051" <- yesterday's finished job
```
Neither of these jobs is active, so the `DataflowJobAlreadyExistsError` rule
doesn't prevent this. The operator then defers on a job that already reached
`JOB_STATE_DONE`, while the process that is still submitting this run's job is
dropped.
Would it make sense to only accept jobs created after this task started, and
keep waiting when every match is terminal?
--
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]
