apache / apache/airflow

DataflowJobStatusTrigger - Add param job_name

Open
#72,875 1 comment 0 reactions 0 assignees View on GitHub
kind:feature provider:google
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

Open the contributing 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

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.