Alwaysgaurav1 opened a new pull request, #69614: URL: https://github.com/apache/airflow/pull/69614
### Description When using `BeamRunPythonPipelineOperator`, `BeamRunJavaPipelineOperator`, or `BeamRunGoPipelineOperator` with the `DataflowRunner`, the operator relies on scraping the launcher's stdout log lines to extract the Dataflow Job ID (using `JOB_ID_PATTERN`). If logging is not configured to include INFO level (e.g., the root logger is left at the default WARNING), the SDK's INFO line indicating the created job ID is never printed to stdout. This causes the job ID extraction to fail silently, leaving `self.dataflow_job_id` as `None`. - In non-deferrable mode, `DataflowHook.wait_for_done` falls back to prefix-based lookup and the task succeeds. - In deferrable mode, the operator defers with `job_id=None` and fails on resume with a `400 Request must contain a job and project id` error. This PR fixes the issue by: 1. Adding a new `fetch_job_id_by_name` method to `DataflowHook`. 2. Implementing a fallback in the Python, Java, and Go operators to resolve the `dataflow_job_id` using the hook method before calling `DataflowJobLink.persist(...)` or deferring. ### Verification - Added unit tests `test_fetch_job_id_by_name` and `test_fetch_job_id_by_name_multiple_jobs` in `test_dataflow.py`. - Verified that all existing unit tests in `test_beam.py` pass successfully (ensuring mock objects in tests are not assigned to the job ID by checking `isinstance(..., str)`). Closes: #68279 -- 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]
