You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "gingeekrishna (via GitHub)" <gi...@apache.org> on 2026/06/28 06:07:31 UTC
[PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
gingeekrishna opened a new pull request, #69102:
URL: https://github.com/apache/airflow/pull/69102
Closes #68279
## Problem
When the Dataflow launcher subprocess runs with the default WARNING log level, it does **not** emit the `"Created job with id: [...]"` line that the Beam operator parses to capture the Dataflow job ID. This leaves `dataflow_job_id = None`.
The previous PRs (#67711, #68720) addressed this by adding a fallback **after** the launcher subprocess finished — but as reviewer @MaksYermak correctly noted, that is not the root cause fix: by the time the launcher exits, the Dataflow job may have already completed, so deferral never gets a chance to free the Airflow worker.
## Root Cause Fix
The correct fix is to capture the job ID **during** the stdout-reading loop, before the launcher finishes, so the operator can truly defer.
### Changes
**`providers/google/.../hooks/dataflow.py`**
- Add `DataflowHook.fetch_job_id_by_name(prefix_name, location, project_id)` — looks up a Dataflow job by name prefix via the API, returning its ID.
**`providers/apache/beam/.../hooks/beam.py`**
- Add `import time`
- Add `periodic_callback: Callable[[], None] | None = None` parameter to `run_beam_command()`, `_start_pipeline()`, `start_python_pipeline()`, and `start_java_pipeline()`
- In `run_beam_command()`: invoke `periodic_callback()` roughly every 5 seconds while the subprocess is running (using `time.monotonic()` tracking). After each periodic call, check `is_dataflow_job_id_exist_callback()` and exit early if the ID has been resolved — before the subprocess finishes.
**`providers/apache/beam/.../operators/beam.py`**
- Add `BeamDataflowMixin.__get_dataflow_job_id_poll_callback()`: returns a closure that calls `DataflowHook.fetch_job_id_by_name()` and sets `self.dataflow_job_id` when a matching job is found; silently retries on transient errors.
- Update `BeamRunPythonPipelineOperator.execute_on_dataflow()` and `BeamRunJavaPipelineOperator.execute_on_dataflow()` to create and pass this callback.
### How this fixes the issue
1. Beam launcher starts (with DataflowRunner) — the Dataflow job is submitted.
2. Every ~5 s, the periodic callback polls the Dataflow Jobs API by job name prefix.
3. Once the job appears, `dataflow_job_id` is set, `is_dataflow_job_id_exist_callback()` returns `True`, and the stdout-reading loop exits **immediately** — before the Dataflow job finishes.
4. The operator defers successfully, freeing the Airflow worker. The Dataflow job continues running on Google Cloud.
This path is the same whether or not the launcher emits a job-ID line to stdout. If stdout does emit the line, `process_line_callback` sets `dataflow_job_id` and the loop exits on the next `is_dataflow_job_id_exist_callback()` check, as before.
## Tests
- Updated hook-level tests to include `periodic_callback=None` in `run_beam_command` mock assertions (all callers that don't pass a periodic_callback).
- Updated operator-level `test_exec_dataflow_runner` tests to include `periodic_callback=mock.ANY`.
- Added `test_exec_dataflow_runner_periodic_callback_fetches_job_id` for both `BeamRunPythonPipelineOperator` and `BeamRunJavaPipelineOperator`: captures the `periodic_callback` passed by the operator, calls it directly, and asserts that `dataflow_job_id` is set by polling `fetch_job_id_by_name`.
## Checklist
- [x] Root cause fixed (not a post-hoc fallback)
- [x] Backward compatible: `periodic_callback` defaults to `None`; existing callers are unaffected
- [x] Go operator not touched (it is sync-only, no `execute_on_dataflow`)
- [x] Syntax validated with `py_compile`
- [x] Newsfragment added: `providers/apache/beam/newsfragments/68279.bugfix.rst`
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "potiuk (via GitHub)" <gi...@apache.org>.
potiuk commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-5021166757
@gingeekrishna This PR has a few issues that need to be addressed before it can be reviewed — please see our [Pull Request quality criteria](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-quality-criteria).
**Issues found:**
- :x: **mypy (type checking)**: Failing: `MyPy providers checks`. Run the relevant mypy hook locally, e.g. `prek run mypy-airflow-core --all-files` (providers: `breeze run mypy path/to/code`).
> **Note:** Your branch is **85 commits behind `main`**. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.
**What to do next:**
- Fix each issue listed above.
- Make sure static checks pass locally (`prek run --from-ref main --stage pre-commit`).
- Mark the PR as "Ready for review" when you're done.
There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the [Airflow Slack](https://s.apache.org/airflow-slack).
---
_Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this [two-stage triage process](https://github.com/apache/airflow/blob/main/contributing-docs/25_maintainer_pr_triage.md#why-the-first-pass-is-automated) so that our maintainers' limited time is spent where it matters most: the conversation with you._
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "MaksYermak (via GitHub)" <gi...@apache.org>.
MaksYermak commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-4874257183
@gingeekrishna sure you can close for not creating confusing noise
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "MaksYermak (via GitHub)" <gi...@apache.org>.
MaksYermak commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-4841618260
> Thanks for the detailed review, @MaksYermak!
>
> We haven't run the full Dataflow system tests — we only exercised unit tests and the existing beam hook test suite locally.
>
> Your point about using the existing `is_dataflow_job_id_exist_callback` to also drive the API poll is a cleaner design than introducing a separate `periodic_callback`, and the note about the fd-blocking concern is well taken.
>
> Since you're close to finishing your own fix with the correct approach, we're happy to close this PR and defer to yours. We'd rather avoid parallel work that heads in incompatible directions. Just let us know if there's anything from this PR you'd find useful to carry forward, or if you'd prefer we simply close it now.
@gingeekrishna here is a PR https://github.com/VladaZakharova/airflow/pull/325 with changes. This PR is blocked by https://github.com/apache/airflow/pull/66952 because in the current time `apache-beam` provider is suspended and firstly we need to un-suspend it for being able to run beam and dataflow unit tests, run dataflow system tests, and developing using breeze.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "henry3260 (via GitHub)" <gi...@apache.org>.
henry3260 commented on code in PR #69102:
URL: https://github.com/apache/airflow/pull/69102#discussion_r3487446968
##########
providers/apache/beam/newsfragments/68279.bugfix.rst:
##########
@@ -0,0 +1 @@
+Fix deferrable ``BeamRunPythonPipelineOperator`` and ``BeamRunJavaPipelineOperator`` failing with ``400 Request must contain a job and project id`` when the Dataflow launcher does not emit a parseable job-ID line to stdout. The operators now poll the Dataflow API periodically while the launcher runs, resolving the job ID by name when stdout does not provide it.
Review Comment:
we don't need this. please remove it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "shahar1 (via GitHub)" <gi...@apache.org>.
shahar1 closed pull request #69102: Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout
URL: https://github.com/apache/airflow/pull/69102
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-4828569085
Thanks for the review, @SameerMesiah97 — good questions. Let me address both:
**1. Worker slot occupancy**
The polling does not add any extra occupancy time. The worker was _already_ occupied for the full duration of the launcher subprocess run before my change — `run_beam_command` has always blocked on `select.select(reads, [], [], 5)` in a loop until the subprocess exits. My `periodic_callback` runs _inside that same existing loop_, within the 5-second `select` timeout that was already there.
If anything, this fix can only _reduce_ occupancy: when the job ID is resolved via the Dataflow API before the launcher finishes printing it to stdout, `run_beam_command` returns early (see line 204) and the operator can defer sooner.
**2. Generic callback vs. dedicated waiter**
The `periodic_callback` is intentionally generic because `run_beam_command` lives in the _Beam_ hook, which should not know about the Dataflow API. A "dedicated waiter" would either need to couple Dataflow API knowledge into that generic hook, or be a wrapper class around `run_beam_command` that ends up doing the same thing less directly.
The callback pattern is already the established idiom here — `process_line_callback` and `is_dataflow_job_id_exist_callback` both pre-existed my change and follow the same shape. `periodic_callback` is just a timed variant of the same mechanism.
Happy to rename it or restructure if you have a specific shape in mind for the dedicated helper.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "gingeekrishna (via GitHub)" <gi...@apache.org>.
gingeekrishna commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-4834281832
Thanks for the detailed review, @MaksYermak!
We haven't run the full Dataflow system tests — we only exercised unit tests and the existing beam hook test suite locally.
Your point about using the existing `is_dataflow_job_id_exist_callback` to also drive the API poll is a cleaner design than introducing a separate `periodic_callback`, and the note about the fd-blocking concern is well taken.
Since you're close to finishing your own fix with the correct approach, we're happy to close this PR and defer to yours. We'd rather avoid parallel work that heads in incompatible directions. Just let us know if there's anything from this PR you'd find useful to carry forward, or if you'd prefer we simply close it now.
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] Fix deferrable Beam Dataflow operators failing with 400 when job ID is missing from stdout [airflow]
Posted by "VladaZakharova (via GitHub)" <gi...@apache.org>.
VladaZakharova commented on PR #69102:
URL: https://github.com/apache/airflow/pull/69102#issuecomment-5636512208
hi @gingeekrishna !
Please, check if you still need this PR or we can close it
--
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: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org