[GitHub] [airflow] mik-laj commented on a change in pull request #9323: Add support for latest Apache Beam SDK in Dataflow operators

[GitBox <gi...@apache.org>]

mik-laj commented on a change in pull request #9323:
URL: https://github.com/apache/airflow/pull/9323#discussion_r440646235



##########
File path: airflow/providers/google/cloud/hooks/dataflow.py
##########
@@ -319,48 +319,50 @@ def __init__(
         on_new_job_id_callback: Optional[Callable[[str], None]] = None
     ) -> None:
         super().__init__()
-        self.log.info("Running command: %s", ' '.join(cmd))
+        self.log.info("Running command: %s", ' '.join(shlex.quote(c) for c in cmd))
         self.on_new_job_id_callback = on_new_job_id_callback
+        self.job_id: Optional[str] = None
         self._proc = subprocess.Popen(
             cmd,
             shell=False,
             stdout=subprocess.PIPE,
             stderr=subprocess.PIPE,
             close_fds=True)
 
-    def _read_line_by_fd(self, fd):
-        if fd == self._proc.stderr.fileno():
-            line = self._proc.stderr.readline().decode()
-            if line:
-                self.log.warning(line[:-1])
-            return line
-
-        if fd == self._proc.stdout.fileno():
-            line = self._proc.stdout.readline().decode()
-            if line:
-                self.log.info(line[:-1])
-            return line
+    def process_fb(self, fd):

Review comment:
       File descriptior
   
   > A file descriptor is either a socket or file object, or a small integer
       gotten from a fileno() method call on one of those.




----------------------------------------------------------------
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.

For queries about this service, please contact Infrastructure at:
users@infra.apache.org