MaksYermak commented on code in PR #72809:
URL: https://github.com/apache/airflow/pull/72809#discussion_r4511489251
##########
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:
@shahar1 I have updated the code
--
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]
