shahar1 commented on code in PR #72809:
URL: https://github.com/apache/airflow/pull/72809#discussion_r4477173294
##########
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)
+
# Waits for Apache Beam pipeline to complete.
log.info("Start waiting for Apache Beam process to complete.")
reads = [fd for fd in (proc.stderr, proc.stdout) if fd is not None]
while True:
# Wait for at least one available fd.
readable_fds, _, _ = select.select(reads, [], [], 5)
- if readable_fds is None:
+ if not readable_fds:
+ if is_dataflow_job_id_exist_callback and
is_dataflow_job_id_exist_callback(
Review Comment:
**major.** The by-name lookup only runs on a 5 s `select` timeout, so a
launcher that exits before the next timeout leaves `dataflow_job_id` as `None`.
A pipeline that calls `p.run()` without `wait_until_finish()`, at the default
WARNING log level, does this: the job is created just before the launcher
exits, and the EOF makes `select` return readable, so no timeout follows the
job's creation.
I ran this function with a silent child that creates the "job" at 6.5 s and
exits 0, and a stub lookup that finds it from then on:
```text
outcome: returned normally after 6.5s
job_id: None lookups: [None]
```
With `deferrable=True`, `execute_on_dataflow` then defers with
`job_id=None`, which is what produces `400 Request must contain a job and
project id.` in #68279. This is also the earlier question on this PR that
wasn't answered: "If the launcher exits without logging a job ID before the
timeout, when would the API lookup run before deferring?"
Two small changes would close it:
1. One last lookup after the launcher exits cleanly, after the final drain:
```python
if proc.returncode == 0 and is_dataflow_job_id_exist_callback:
is_dataflow_job_id_exist_callback(checking_by_job_name=True)
```
2. `if self.deferrable and self.dataflow_job_id:` in both
`execute_on_dataflow` methods. If the job still isn't found (it already
finished, so `filter="ACTIVE"` misses it, or the lookup failed, or the google
provider is older), the task then falls through to the by-name `wait_for_done`
that the sync path already uses, instead of deferring and failing.
##########
providers/apache/beam/src/airflow/providers/apache/beam/hooks/beam.py:
##########
@@ -142,7 +142,7 @@ def process_fd(
func_log = fd_to_log[fd]
for line_raw in iter(fd.readline, b""):
- line = line_raw.decode()
+ line = line_raw.decode(errors="replace")
Review Comment:
**minor.** The earlier review called a split line's missed regex mostly
harmless, but the Java alternative of `JOB_ID_PATTERN` (`Submitted job:
(?P<job_id_java>[^\"\n\s]*)`) has no closing delimiter, so a split line can
match a truncated ID instead of missing. A child writing `Submitted job:
2026-11-0`, pausing, then writing the rest gives:
```text
outcome: returned normally after 0.0s
job_id: '2026-11-0'
```
The task then monitors an ID that doesn't exist. This needs the launcher to
emit the line in more than one write, and I haven't checked whether the Beam
Java launcher ever does, so it's likely rare. Passing only complete lines to
the callback (`if process_line_callback and line.endswith("\n"):`) would leave
this case to the by-name lookup.
##########
providers/apache/beam/src/airflow/providers/apache/beam/operators/beam.py:
##########
@@ -155,8 +155,25 @@ 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
+ and hasattr(self.dataflow_hook, "get_job_id_of_running_job")
+ ):
+ try:
+ self.dataflow_job_id =
self.dataflow_hook.get_job_id_of_running_job(
Review Comment:
**minor.** On the open thread about `DataflowJobAlreadyExistsError`: I think
both descriptions are right, and which one you get depends on timing. With a
stub lookup that returns another run's active job:
```text
launcher fails at t=2s: raised AirflowException: Apache Beam process failed
with return code 1
launcher fails at t=7s: returned normally after 5.0s, job_id 'job-other-run'
```
So the task fails, matching what you reported, when the launcher reaches the
error before 5 s of silence. When staging stays silent for longer, the lookup
wins: the task follows the other run's job and reports its result, and
`on_kill` would cancel that job. This needs `append_job_name=False`. The
earlier suggestion still applies: add a sentence to the `.. warning::`, or
bound the lookup by `createTime`.
##########
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)
+
# Waits for Apache Beam pipeline to complete.
log.info("Start waiting for Apache Beam process to complete.")
reads = [fd for fd in (proc.stderr, proc.stdout) if fd is not None]
while True:
# Wait for at least one available fd.
readable_fds, _, _ = select.select(reads, [], [], 5)
- if readable_fds is None:
+ if not readable_fds:
+ if is_dataflow_job_id_exist_callback and
is_dataflow_job_id_exist_callback(
+ checking_by_job_name=True
Review Comment:
**minor.** `is_dataflow_job_id_exist_callback` is a public parameter of
`BeamHook.start_python_pipeline` and `start_java_pipeline`, previously typed
`Callable[[], bool]`. A caller passing its own zero-argument callback now fails
at the first 5 s of silence:
```text
TypeError: zero_arg_callback() got an unexpected keyword argument
'checking_by_job_name'
```
Nothing in this repository passes one, so this only affects outside callers.
A separate optional callback for the lookup would keep the existing signature
and let the type go back from `Callable[..., bool]`.
##########
providers/apache/beam/tests/unit/apache/beam/hooks/test_beam.py:
##########
@@ -441,6 +443,42 @@ def test_beam_wait_for_done_logging(self, mock_select,
mock_popen, caplog):
assert "apache-beam-stderr-3" in warn_messages
assert "apache-beam-other-stderr" in warn_messages
+ @mock.patch("os.set_blocking")
+ @mock.patch("subprocess.Popen")
+ @mock.patch("select.select")
+ def test_run_beam_command_checks_job_id_by_name_on_timeout(
+ self, mock_select, mock_popen, mock_set_blocking
+ ):
+ fake_logger = MagicMock()
Review Comment:
**minor.** From `.github/instructions/code-review.instructions.md`:
> **Flag any changed or added behaviour without a corresponding test, and
flag tests that a reviewer cannot fail by reverting the PR's change.**
Two changes have no test yet: `errors="replace"` in `process_fd` (no test
feeds it undecodable bytes), and the false branch of the `hasattr` guard (a
`MagicMock` hook always has the attribute).
> **Flag any `mock.Mock()` or `mock.MagicMock()` without `spec` or
`autospec`.**
This applies to the new `fake_logger`, `mock_proc`, `fake_stderr_fd`,
`fake_stdout_fd` and `is_job_id_exist_callback` here.
--
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]
