DataflowJobStatusTrigger - Add param job_name
- Dominant language
- Python
- Stars
- 46.9k
- Forks
- 17.8k
- Avg merge
- 2d 9h
- Merged PRs (30d)
- 472
Description
### Description
Add support for filtering/monitoring Dataflow jobs by `job_name` in `DataflowJobStatusSensor` (and its corresponding trigger `DataflowJobStatusTrigger`), rather than requiring `job_id`.
Currently, `DataflowJobStatusSensor` strictly requires `job_id`. Because Dataflow job names are not globally unique, the sensor should look up jobs matching `job_name` within the specified project/location and target the latest job based on its `createTime` / `startTime`.
### Use case/motivation
In many CI/CD and production environments:
1. Pipelines are triggered externally or via tools that do not automatically pass the generated Dataflow `job_id` downstream via XCom. (I use cloud run job operator to launch dataflow in a java environment that I can manage)
2. Users follow predictable naming conventions (e.g., `daily-etl-pipeline` or `daily-etl-pipeline-{{ ds_nodash }}`).
3. Needing `job_id` forces users to write custom Python operators or custom hooks just to fetch the job ID before handing it over to `DataflowJobStatusSensor`.
Allowing `job_name` directly in `DataflowJobStatusSensor` will simplify DAGs and make the sensor consistent with other operators (like `DataflowStopJobOperator`) that already allow filtering by name.
## Proposition
- Update `DataflowJobStatusSensor` arguments to make `job_id: str | None = None` and add `job_name: str | None = None` (raising an `AirflowException` if neither or both are provided).
- In `poke()` and `DataflowJobStatusTrigger`:
- If `job_name` is provided, call `DataflowHook.list_jobs()` (or equivalent API) filtered by location/project.
- Filter jobs matching `job_name` and sort by `createTime` descending.
- Track the status of the newest job.
- If no job matching `job_name` is found yet, continue sensing until timeout or until the job appears.
### Related issues
_No response_
### Are you willing to submit a PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
Contributor guide
Research direction
Start by locating DataflowJobStatusSensor and DataflowJobStatusTrigger, then inspect DataflowHook.list_jobs() and the existing job_id handling. Implement the mutually exclusive job_id/job_name behavior, newest matching job selection, and continued sensing when no named job exists; done means both sensor and trigger support the requested filtering and status tracking.
Written by the indexing model from the issue text.
Assessment
- Tech stack
- python
- Domain
- data-engineering
- Issue type
- Feature
- Difficulty
- 4/5
- Estimated time
- 3-5 days
- Activity status
- Active
- Clarity
- Mostly clear
- Newbie friendliness
- 65/100