Add job_name support to DataflowJobStatusSensor and DataflowJobStatusTrigger - #72879
bujjibabukatta wants to merge 3 commits into
Conversation
|
Thanks for working on this feature. I reviewed the implementation and found a few areas that may be worth addressing before merge: sync name lookup should avoid the private _fetch_all_jobs() method, async lookup should handle pagination when selecting the newest job, and the trigger should avoid mutating its serialized job_id state. I also found a few useful regression cases around job_id/job_name exclusivity, newest-match selection, no-match behavior, and trigger serialization. I have a tested version of these changes locally with the relevant Dataflow tests passing. Happy to share the patch/commit if useful. |
| matching_jobs = [job for job in all_jobs if job.get("name") == job_name] | ||
| if not matching_jobs: | ||
| return None | ||
| matching_jobs.sort(key=lambda job: job.get("createTime", ""), reverse=True) |
There was a problem hiding this comment.
How do we know this is the right way to do this? In the fetch_job_by_id counterpart, this is the code that is used:
return (
self._dataflow.projects()
.locations()
.jobs()
.get(
projectId=self._project_number,
location=self._job_location,
jobId=job_id,
)
.execute(num_retries=self._num_retries)
)| if not matching_jobs: | ||
| return None | ||
| matching_jobs.sort(key=lambda job: job.create_time, reverse=True) | ||
| return matching_jobs[0] |
There was a problem hiding this comment.
Same comment as above, "brute-force" selecting the first matching job doesn't feel right.
|
Hi @jroachgolf84 and @Kunal8954, thanks for the review! I've pushed fixes for the points raised: Renamed _fetch_all_jobs() to public fetch_all_jobs() and updated its caller On the "brute-force" listing concern — Dataflow's API has no get-by-name endpoint, only get-by-ID or list, so client-side filtering is the only option I'm aware of. Open to suggestions if there's a better approach. Thanks again for the time you both put into this! |
jroachgolf84
left a comment
There was a problem hiding this comment.
I think you'll want to add tests for much of the logic that you've defined in your hooks and sensors. Tests are only updated right now for triggers.
| location=location, | ||
| ) | ||
| all_jobs = jobs_controller.fetch_all_jobs() | ||
| matching_jobs = [job for job in all_jobs if job.get("name") == job_name] |
There was a problem hiding this comment.
Does case matter here? i.e. should MY_JOB == my_job == My_Job?
| :return: the most recent matching Job, or None if no job matches. | ||
| """ | ||
| page_result = await self.list_jobs(project_id=project_id, location=location) | ||
| matching_jobs = [job async for job in page_result if job.name == job_name] |
There was a problem hiding this comment.
Same question here about case.
|
Hello @bujjibabukatta - 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 16 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 |
Summary
DataflowJobStatusSensor (and its trigger, DataflowJobStatusTrigger) previously
required a job_id, forcing users who don't have the job ID readily available
(e.g. jobs launched by external tools that don't push it to XCom) to write
custom operators just to look it up. This PR adds an alternative job_name
parameter so the sensor/trigger can locate and monitor a job by name instead.
Root cause
Dataflow job names are not globally unique, so simply exposing job_name as a
drop-in replacement for job_id isn't safe — multiple jobs (across retries/
backfills) can share a name. The sensor and trigger had no mechanism to
disambiguate between same-named jobs or to handle the case where the named
job hasn't been created yet.
Fix
job_name (exactly one required, enforced via exactly_one(), raising
AirflowException otherwise).
AsyncDataflowHook.get_job_by_name() (async), which list jobs in the given
project/location, filter to exact name matches, and return the one with
the latest createTime/create_time.
and the trigger's run() sleeps and retries, so both keep sensing until the
job appears or the task times out.
is unambiguous.
(e.g. job_name="daily-etl-{{ ds_nodash }}").
closes: #72875
Was generative AI tooling used ?
Generated-by: Claude following the guidelines