As discussed in  https://github.com/apache/airflow/pull/15534#discussion_r620500075, we could possibly rewrite Beam Hook to remove the need of storing state in it.
"""
        Prints output to logs.
        :param fd: File descriptor.
        """
        if fd not in (self._proc.stdout, self._proc.stderr):
            raise Exception("No data in stderr or in stdout.")
        fd_to_log = {self._proc.stderr: self.log.warning, self._proc.stdout: self.log.info}
        func_log = fd_to_log[fd]
        while True:
            line = fd.readline().decode()
            if not line:
                return
            if self.process_line_callback:
                self.process_line_callback(line)
            func_log(line.rstrip("\n"))
    def wait_for_done(self) -> None: