MaksYermak commented on PR #72809: URL: https://github.com/apache/airflow/pull/72809#issuecomment-6124043000
Hello @potiuk, thank you for the review! Honestly, I do not fully understand your points and I need clarification. Could you please clarify for me? > 1. **The "running job" lookup can return a finished job.** `get_job_id_of_running_job` → `_get_current_jobs` matches by name _prefix_ and then takes the single match regardless of state, or, among several, the newest by `createTime` even if it's `JOB_STATE_DONE`. With `append_job_name=False` and a silent launcher (the exact case this PR targets), the first lookup fires 5 s after start, usually before staging has created the new job. On the second run of such a Dag the previous DONE job is returned, the operator defers on it and the task succeeds while the real job runs unmonitored. Your manual test had stderr warnings delaying the first timeout until the job was already PENDING, so it doesn't cover this. Please restrict this path to non-terminal jobs with an exact name match (`jobs.list` with `filter=ACTIVE` also makes it much cheaper), and keep the "newest terminal job" fallback out of the stop-reading decision. This point based on the `5s` timeout in this line of code `readable_fds, _, _ = select.select(reads, [], [], 5)` and in this point you mentioned that the code checked `dataflow_job_id` before runs started in this `5s` delay. I do not clearly understand this assumption because for `Dataflow` job we might not have `fd`s as user can disable logs in his `Pipeline`s code. As a result the code will not write anything in `stdout` and `stderr` and we do not have any `fd`s. It was the cause of bug which I have tried to solve. Because in case when user do not enable logs in the pipeline code we do not have `dataflow_job_id`. It is a reason why I have added the `if-cause` code in this `5s` delay otherwise the code will be in the infinite loop till process will not die. > 2. **Cross-provider compatibility.** The beam operator calls `DataflowHook.get_job_id_of_running_job`, which this PR adds to the google provider, so it will only exist from the next google release. Beam's `google` extra doesn't depend on the google provider at all (it only lists `apache-beam[gcp]`), so new beam plus an already-installed google provider fails with `AttributeError` after 5 s of launcher silence. > Please declare the dependency in `providers/apache/beam/pyproject.toml`, in the `google` extra, with the special `# use next version` marker: > ```toml > "google" = [ > "apache-beam[gcp]>=2.76.0", > "apache-airflow-providers-google>=22.6.0", # use next version > ] > ``` > > > > > > > > > > > > Keep the version at the current one (22.6.0) — don't bump it yourself. The exact `# use next version` comment tells the release tooling to raise it to the google version being released alongside this beam release, i.e. the first one that has `get_job_id_of_running_job`, and the Release Manager releases both together. See the `# use next version` explanation in `contributing-docs/13_airflow_dependencies_and_extras.rst`. Alternatively, guard the call (e.g. `hasattr(self.dataflow_hook, "get_job_id_of_running_job")`) and fall back to the current behaviour, but the dependency marker is the cleaner fix. It makes sense for me and I am ready to update the code. One question from my side. Do we need a binding with google provider in apache beam requirements? As I see in this [PR](https://github.com/apache/airflow/pull/54982) previously we decided to add logic for checking presence of google provider not added it as a requirements to beam provider. > 3. **Tests for the beam side.** Nothing tests the timeout → API-lookup path in `run_beam_command`, the `os.set_blocking` change, the operator callback, or `get_job_id_of_running_job` / `fetch_job_id`. Each changed behaviour needs a test that fails without the change. (Please also mock `os.set_blocking` there; with a `MagicMock` Popen it currently ends up calling `set_blocking(1, False)` on the test process's stdout.) Could you please clarify what timeout are you mentioned? In this code I see `5s` timeout in `readable_fds, _, _ = select.select(reads, [], [], 5)` line which is part of `select` library. Should I create a unit test for this timeout or for something else? > 4. **Non-blocking reads for all runners.** `process_fd` decodes whatever `readline()` returns. In non-blocking mode that can be a partial line, so a split UTF-8 character raises `UnicodeDecodeError`, and the job-id regex can miss an ID split across reads. Please decode defensively (or buffer up to newline) and consider limiting the non-blocking switch to the Dataflow path. This suggestion looks like changes in `process_fd` function which I did not touch in this code. I am not sure that this PR is a good place for these changes. Maybe we can create another PR for this, WDYT? > 5. **Lookup robustness.** The lookup runs on every 5 s of silence and lists all jobs in the region each time. If it raises (e.g. missing `dataflow.jobs.list` permission) the task fails while the launcher keeps running. Please catch and log the error, then continue reading. > I need a clarification about what `5s` lookup are we talking about? Because, I see `5s` only in this line `readable_fds, _, _ = select.select(reads, [], [], 5)`. Am I correct understand that this point about calling `is_dataflow_job_id_exist_callback` inside `while True` loop? > Smaller points: the new `.. warning::` says the subprocess is _terminated_ and only in deferrable mode, but the code just stops reading it and also returns early in non-deferrable mode. Also, `fetch_job_id` / `get_job_id_of_running_job` could use `-> str | None` annotations. > I will update `warning` and add annotation. -- 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]
