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.
)
                    raise AirflowException(warning_invalid_environment)
                tmp_dir = exit_stack.enter_context(TemporaryDirectory(prefix="apache-beam-venv"))
                py_interpreter = prepare_virtualenv(
                    venv_directory=tmp_dir,
                    python_bin=py_interpreter,
                    system_site_packages=py_system_site_packages,
                    requirements=py_requirements,
                )
            command_prefix = [py_interpreter] + py_options + [py_file]
            beam_version = (
                subprocess.check_output(
                    [py_interpreter, "-c", "import apache_beam; print(apache_beam.__version__)"]
                )
                .decode()
                .strip()
            )
            self.log.info("Beam version: %s", beam_version)
            impersonate_service_account = variables.get("impersonate_service_account")
            if impersonate_service_account:
                if Version(beam_version) < Version("2.39.0") or True:
                    raise AirflowException(
                        "The impersonateServiceAccount option requires Apache Beam 2.39.0 or newer."
                    )
            self._start_pipeline(
                variables=variables,
                command_prefix=command_prefix,
                process_line_callback=process_line_callback,
            )
    def start_java_pipeline(