potiuk commented on code in PR #72809:
URL: https://github.com/apache/airflow/pull/72809#discussion_r4422010578
##########
providers/apache/beam/src/airflow/providers/apache/beam/hooks/beam.py:
##########
@@ -176,13 +176,22 @@ def run_beam_command(
stderr=subprocess.PIPE,
close_fds=True,
)
+
+ if is_dataflow_job_id_exist_callback and hasattr(os, "set_blocking") and
proc.stdout and proc.stderr:
+ os.set_blocking(proc.stdout.fileno(), False)
+ os.set_blocking(proc.stderr.fileno(), False)
Review Comment:
Switching these pipes to non-blocking changes what `process_fd` receives. In
blocking mode `fd.readline()` only returns a complete line (or `b""` at EOF).
In non-blocking mode, `BufferedReader.readline()` returns whatever bytes are
available when the raw read hits `EAGAIN`, so a line can arrive in two pieces.
If the split falls inside a multi-byte UTF-8 character (any non-ASCII text in
Beam/Dataflow output: user log messages, file paths, localized warnings), `line
= line_raw.decode()` at L146 raises `UnicodeDecodeError` and the task fails.
That's why the fix belongs in this PR even though L146 isn't in the diff:
```python
# in process_fd (L146)
line = line_raw.decode(errors="replace")
```
(A split line can also make `process_line_callback` miss the job-id regex
for that read. That's mostly harmless now that the name lookup is a fallback,
but it's another reason to keep this to the Dataflow path, as you've done.)
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -339,6 +339,15 @@ def _get_current_jobs(self) -> list[dict]:
jobs = self._fetch_jobs_by_prefix_name(self._job_name.lower())
if len(jobs) == 1:
self._job_id = jobs[0]["id"]
+ elif len(jobs) > 1 and not self._multiple_jobs:
+ active_jobs = [
+ job for job in jobs if job.get("currentState") not in
DataflowJobStatus.TERMINAL_STATES
+ ]
+ if len(active_jobs) == 1:
+ self._job_id = active_jobs[0]["id"]
+ else:
+ jobs.sort(key=lambda j: j.get("createTime", ""),
reverse=True)
+ self._job_id = jobs[0]["id"]
Review Comment:
`jobs` here comes from `_fetch_jobs_by_prefix_name`, so it holds every job
whose name *starts with* `self._job_name`. Picking "the single active one, else
the newest" can therefore select a different job: for `daily-etl` with
`append_job_name=False`, an active `daily-etl-backfill` wins over this run's
own job. If none is active, the newest terminal job is chosen, which may also
be a sibling or a stale run.
This method backs `wait_for_done(job_name=..., job_id=None)` for every
Dataflow operator, not only beam, and the beam fix now goes through
`fetch_job_id` instead. Could you drop this hunk? If it's needed for another
case, please match exact names (`job.get("name") == self._job_name.lower()`)
and explain that case in the PR.
##########
providers/apache/beam/src/airflow/providers/apache/beam/operators/beam.py:
##########
@@ -155,8 +155,20 @@ def set_current_dataflow_job_id(job_id):
on_new_job_id_callback=set_current_dataflow_job_id
)
- def __is_dataflow_job_id_exist_callback(self) -> Callable[[], bool]:
- def is_dataflow_job_id_exist() -> bool:
+ def __is_dataflow_job_id_exist_callback(self) -> Callable[..., bool]:
+ def is_dataflow_job_id_exist(checking_by_job_name: bool = False) ->
bool:
+ if checking_by_job_name and self.dataflow_hook:
+ try:
+ self.dataflow_job_id =
self.dataflow_hook.get_job_id_of_running_job(
+ job_name=self.dataflow_job_name, # type: ignore
+ project_id=self.dataflow_config.project_id,
+ location=self.dataflow_config.location,
+ )
+ except Exception:
+ self.log.warning( # type: ignore
+ "Failed to get Dataflow job ID by job name '%s'",
+ self.dataflow_job_name, # type: ignore
+ )
Review Comment:
With beam installed alongside a google provider that predates
`get_job_id_of_running_job`, this call raises `AttributeError`. The `except
Exception` catches it, so the task no longer fails, but it logs "Failed to get
Dataflow job ID by job name …" every 5 s of launcher silence with no indication
why. Suggest:
```python
if (
checking_by_job_name
and self.dataflow_hook
and hasattr(self.dataflow_hook, "get_job_id_of_running_job")
):
try:
...
except Exception:
self.log.warning("Failed to get Dataflow job ID by job name '%s'",
self.dataflow_job_name, exc_info=True)
```
The `hasattr` guard keeps the old behaviour on an older google provider (the
same approach as #54982, so no hard dependency is needed), and `exc_info=True`
shows the cause of a real failure, e.g. a missing `dataflow.jobs.list`
permission.
##########
providers/google/src/airflow/providers/google/cloud/hooks/dataflow.py:
##########
@@ -434,13 +443,40 @@ def fetch_job_autoscaling_events_by_id(self, job_id: str)
-> list[dict]:
autoscaling_events.extend(response.get("autoscalingEvents", []))
return autoscaling_events
- def _fetch_all_jobs(self) -> list[dict]:
- request = (
- self._dataflow.projects()
- .locations()
- .jobs()
- .list(projectId=self._project_number, location=self._job_location)
- )
+ def fetch_job_id(self) -> str | None:
+ if self._job_id:
+ return self._job_id
+ if not self._job_name:
+ raise ValueError("Missing both dataflow job ID and name.")
+ jobs = self._fetch_all_jobs(jobs_filter="ACTIVE")
+ job_name = self._job_name.lower()
+ active_jobs = [
+ job
+ for job in jobs
+ if job.get("name") == job_name
+ and job.get("currentState") not in
DataflowJobStatus.TERMINAL_STATES
+ ]
Review Comment:
Minor, not blocking. With `append_job_name=False`, if an earlier try's job
is still running under the same name (worker crash plus retry, or a streaming
job), the first 5 s lookup returns *that* job, and the operator then monitors
or defers on it. The new launcher, meanwhile, fails with
`DataflowJobAlreadyExistsError`, and since we've stopped reading the subprocess
nobody sees it; before this PR the task would fail. A `createTime` lower bound
(task start time) would close this. Otherwise a sentence in the `.. warning::`
docstring would do.
--
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]
