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.
"""
    Class responsible for running pipeline command in subprocess
    :param cmd: Parts of the command to be run in subprocess
    :param process_line_callback: Optional callback which can be used to process
        stdout and stderr to detect job id
    :param working_directory: Working directory
    """
    def __init__(
        self,
        cmd: list[str],
        process_line_callback: Callable[[str], None] | None = None,
        working_directory: str | None = None,
    ) -> None:
        super().__init__()
        self.log.info("Running command: %s", " ".join(shlex.quote(c) for c in cmd))
        self.process_line_callback = process_line_callback
        self.job_id: str | None = None
        self._proc = subprocess.Popen(
            cmd,
            cwd=working_directory,
            shell=False,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            close_fds=True,
        )
    def _process_fd(self, fd):