namanjain24-sudo opened a new pull request, #73356:
URL: https://github.com/apache/airflow/pull/73356
`DataflowJobStatusSensor` and `DataflowJobStatusTrigger` can only watch a
job by id. When the job is launched outside the Dag, as in #72875 where it is
started from a Cloud Run job, the id is never available downstream. This adds a
`job_name` parameter to both, mutually exclusive with `job_id`.
How a name is resolved:
- Dataflow job names are not unique over time, so the new
`DataflowHook.get_latest_job_by_name` and its async counterpart match the name
exactly and return the most recent job with that name, as the issue proposes.
This is deliberately not the prefix match that
`DataflowStopJobOperator.job_name_prefix` does.
- The match happens client side, the same way `_fetch_jobs_by_prefix_name`
in this hook already works. Newer API clients do accept a `name` argument on
`jobs.list`, but the minimum versions this provider supports do not:
`google-api-python-client` 2.0.2's bundled discovery document and
`ListJobsRequest` in `google-cloud-dataflow-client` 0.8.6 have no such field.
Switching to the server-side filter would need both floors raised, and I can do
that here or as a follow-up if preferred.
- `jobs.list` only promises the summary view, which documents a start time
but not a create time. So the ordering key is `createTime`, then `startTime`,
then the job that is not in a terminal state (only one active job per name can
exist in a project and region). Ordering by `createTime` alone made the sync
path return whichever job the API listed first, and made the async path raise
`TypeError` comparing two unset timestamps. Both cases have tests.
- While no job with the name exists yet, the sensor keeps poking and the
trigger keeps polling, so the Dag can start waiting before the job is created.
A finished job from an earlier run with the same name also counts, so the docs
suggest making reused names unique per run, e.g. `daily-etl-{{ ds_nodash }}`.
Compatibility: `job_name` is appended last in the trigger signature, so
existing positional calls keep working, and a serialized trigger from before
this change (no `job_name` key) still deserializes. Log messages and the
terminal-state error now report the id of the job that was actually resolved.
Tests:
- 24 new tests across the hook, sensor and trigger. With the three source
files reverted to main, every one of them fails or errors.
- The Dataflow hook, sensor, trigger and operator tests (284 passed) and the
Google sensors and triggers suites (607 passed, 6 skipped) pass locally on
current main. mypy is clean on the three changed modules, and the
templated-fields check passes.
closes: #72875
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: a Gen-AI coding assistant, following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions).
I reviewed the change and ran the checks above.
--
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]
