gingeekrishna commented on issue #68279: URL: https://github.com/apache/airflow/issues/68279#issuecomment-4825083334
I've submitted a fix in PR #69102: https://github.com/apache/airflow/pull/69102 ## Root Cause The previous PRs (#67711, #68720) both added a fallback **after** the launcher subprocess exited — but as @MaksYermak correctly noted, that doesn't fix the real problem. By the time the launcher finishes, the Dataflow job may have already completed, so there's nothing left to defer. The actual root cause is that the stdout-reading loop in `run_beam_command()` runs until the launcher process exits. If the launcher never emits the `"Created job with id: [...]"` line (which requires INFO logging, not the default WARNING), `dataflow_job_id` stays `None` and deferral fails. ## What the Fix Does The fix captures the job ID **during** the stdout loop, before the launcher finishes: 1. **`DataflowHook.fetch_job_id_by_name()`** — new method that polls the Dataflow Jobs API by job name prefix to retrieve the job ID. 2. **`run_beam_command()` — new `periodic_callback` parameter** — invoked roughly every 5 seconds while the launcher is running (tracked with `time.monotonic()`). After each call, `is_dataflow_job_id_exist_callback()` is checked; if the ID has been resolved, the loop exits immediately — before the Dataflow job completes. 3. **`BeamDataflowMixin.__get_dataflow_job_id_poll_callback()`** — builds the closure that calls `fetch_job_id_by_name()` and sets `self.dataflow_job_id`. Both `BeamRunPythonPipelineOperator` and `BeamRunJavaPipelineOperator` now pass it when running on DataflowRunner. This way, even if the launcher's stdout produces no job-ID line at all, the operator finds the job ID via the API within ~5 seconds of submission — early enough to defer and free the Airflow worker while the job runs on Google Cloud. -- 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]
