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.
self,
        runner: str,
    ) -> None:
        self.runner = runner
        super().__init__()
    def _start_pipeline(
        self,
        variables: dict,
        command_prefix: list[str],
        process_line_callback: Callable[[str], None] | None = None,
        working_directory: str | None = None,
    ) -> None:
        cmd = command_prefix + [
            f"--runner={self.runner}",
        ]
        if variables:
            cmd.extend(beam_options_to_args(variables))
        cmd_runner = BeamCommandRunner(
            cmd=cmd,
            process_line_callback=process_line_callback,
            working_directory=working_directory,
        )
        cmd_runner.wait_for_done()
    def start_python_pipeline(