ashokballolli opened a new pull request, #77069: URL: https://github.com/apache/airflow/pull/77069
`DataflowJobStatusSensor` works only with `job_id`. But when the Dataflow job is started outside Airflow (like from a Cloud Run job, as mentioned in the issue), the Dag does not get the job ID. Mostly we know the job name because it follows some naming pattern, so it is better if the sensor can find the job using the name. This PR adds `job_name` to the sensor and to `DataflowJobStatusTrigger`. Only one of `job_id` or `job_name` can be given. When `job_name` is used, on every check the jobs in that location are listed and the latest created job with the exact same name is taken, in any state. If no job with that name exists yet, the sensor (or the trigger in deferrable mode) keeps waiting. Nothing changes for `job_id`. Few points about the approach: - Dataflow API does not have any call to get a job by name, so listing the jobs is the only way. `DataflowHook.get_job_id_of_running_job` is also doing the same thing. I added public methods `DataflowHook.get_latest_job_by_name` and `AsyncDataflowHook.get_latest_job_by_name` for this, so the sensor is not calling private methods of the hook. The async one goes through all the pages of the response. - If more than one job has the same name, the newest one by `createTime` is taken, not the first one in the list. - The name is converted to lowercase before matching, same as the existing name lookup in the hook. Dataflow allows only lowercase job names, so `My-Job` can only mean `my-job`. - The trigger does not save the job ID it found, so it always checks the latest job with that name, same as the sensor. - If the same name is used for every run, the sensor can pick up the previous run before the new run starts. I added a note in the docs to add something unique like the date in the name. I have not added any check for this in the code as the issue did not ask for it, but I can add it if needed. #72879 and #73356 were closed earlier without merging. This PR also takes care of the review comments given on #72879 (no call to private `_fetch_all_jobs` from the sensor, pagination in the async lookup, trigger not changing its own state, case handling, and tests for the hooks and sensor). What I checked: - Unit tests for the Dataflow hook, sensor and trigger: all 267 passed. I also ran the 16 new tests without the change and all of them fail, so they are really testing the new code. - Existing `job_id` tests are not changed and still pass, so nothing changes for current users. - `job_name` is added at the end of the trigger arguments with default `None`, so a trigger saved by an older provider version (without `job_name`) still loads after upgrade. - For the async lookup, I checked the `ListJobsAsyncPager` code in `google-cloud-dataflow-client` to confirm that `async for` goes through all the pages. - `prek` static checks on the changed files passed (ruff, template fields check, provider checks). - Not tested on a real Dataflow project yet. closes: #72875 --- ##### Was generative AI tooling used to co-author this PR? - [X] Yes: Claude Code (Opus 5.5) I work as a data engineer on GCP, mostly with Airflow on Cloud Composer. I used the tool to help me go through this part of the Google provider code and with writing the tests. I went through all the changes, ran the tests and checks myself. Happy to make any changes based on review. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
