Repository navigation
Allow DataflowJobStatusSensor to wait on a job name - #73356
namanjain24-sudo wants to merge 2 commits into
Conversation
boring-cyborg
Bot
added
area:providers
kind:documentation
provider:google
DataflowJobStatusSensor and DataflowJobStatusTrigger could only watch a job by id, so a DAG that did not launch the job itself had no way to wait for it. Add a job_name parameter to both, mutually exclusive with job_id. Dataflow job names are not unique over time, so the new hook methods DataflowHook.get_latest_job_by_name and its async counterpart match the name exactly and return the most recent job with that name. This is deliberately not the prefix match DataflowStopJobOperator.job_name_prefix uses, because the two parameters mean different things. The Dataflow REST API filters jobs.list by job state rather than by name, so the match has to happen client side, the same way _fetch_jobs_by_prefix_name already does it. Picking the most recent one cannot rely on createTime alone. The list method promises only the summary view, which documents a start time but not a create time, so the ordering key falls back to startTime, and then to the job that has not reached a terminal state, since only one active job with a given name can exist in a project within one region at a time. Without the fallbacks the synchronous path silently returned whichever job the API happened to list first, and the asynchronous path raised a TypeError comparing two unset timestamps. While no job with that name exists yet the sensor keeps poking and the trigger keeps polling, so a DAG can start waiting before the job is created. Log messages and the terminal-state error now report the id of the job that was actually resolved.
The sensor picks the most recently created job with the given name, so a job left over from an earlier run with the same name counts as well. Say so in the docstring and the how-to guide, and suggest making the name unique per run.
e663af7 to
56f91de
Compare
potiuk
added
the
closed because of open PR limit
|
Hello @namanjain24-sudo - thank you for your contributions to Apache Airflow! The Airflow community has introduced a limit of 5 open pull requests at a time for contributors without write access to the repository. You currently have 7 open pull requests, so - as a one-time step of introducing the limit - we closed the ones where maintainers have not engaged yet:
These pull requests stay open because maintainers are already engaged in them - they count towards your limit:
This is not a judgement of you or of your changes. We never told contributors before that opening many pull requests at once was a problem, so there is nothing to feel bad about - and nothing is lost: your branches, commits and the review history stay where they are. What we ask you to do is to make your first prioritization decision: choose which of the pull requests above matter most to you, and reopen them (up to 5 open at a time, including the ones still open) with the "Reopen pull request" button or While your pull requests are waiting for review, the most valuable thing you can do is help in other ways - reviewing other contributors' pull requests, helping with issues, and taking part in the discussions on the devlist and Slack. Why we introduced the limit, what it means for you and how to reopen or restore a pull request is explained in https://github.com/apache/airflow/blob/main/contributing-docs/32_open_pull_request_limit.rst. Drafted-by: Claude Code (Opus 5); reviewed by @potiuk before posting |
DataflowJobStatusSensorandDataflowJobStatusTriggercan 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 ajob_nameparameter to both, mutually exclusive withjob_id.How a name is resolved:
DataflowHook.get_latest_job_by_nameand 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 thatDataflowStopJobOperator.job_name_prefixdoes._fetch_jobs_by_prefix_namein this hook already works. Newer API clients do accept anameargument onjobs.list, but the minimum versions this provider supports do not:google-api-python-client2.0.2's bundled discovery document andListJobsRequestingoogle-cloud-dataflow-client0.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.listonly promises the summary view, which documents a start time but not a create time. So the ordering key iscreateTime, thenstartTime, then the job that is not in a terminal state (only one active job per name can exist in a project and region). Ordering bycreateTimealone made the sync path return whichever job the API listed first, and made the async path raiseTypeErrorcomparing two unset timestamps. Both cases have tests.daily-etl-{{ ds_nodash }}.Compatibility:
job_nameis appended last in the trigger signature, so existing positional calls keep working, and a serialized trigger from before this change (nojob_namekey) still deserializes. Log messages and the terminal-state error now report the id of the job that was actually resolved.Tests:
closes: #72875
Was generative AI tooling used to co-author this PR?
Generated-by: a Gen-AI coding assistant, following the guidelines. I reviewed the change and ran the checks above.