DataflowJobStatusTrigger - Add param job_name #72875

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

• I agree to follow this project's Code of Conduct