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.
"""Waits for Apache Beam pipeline to complete."""
        self.log.info("Start waiting for Apache Beam process to complete.")
        reads = [self._proc.stderr, self._proc.stdout]
        while True:
            
            readable_fds, _, _ = select.select(reads, [], [], 5)
            if readable_fds is None:
                self.log.info("Waiting for Apache Beam process to complete.")
                continue
            for readable_fd in readable_fds:
                self._process_fd(readable_fd)
            if self._proc.poll() is not None:
                break
        
        for readable_fd in reads:
            self._process_fd(readable_fd)
        self.log.info("Process exited with return code: %s", self._proc.returncode)
        if self._proc.returncode != 0:
            raise AirflowException(f"Apache Beam process failed with return code {self._proc.returncode}")
class BeamHook(BaseHook):
    """
    Hook for Apache Beam.
    All the methods in the hook where project_id is used must be called with
    keyword arguments rather than positional.
    :param runner: Runner type
    """
    def __init__(