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.
DataflowHook,
    DataflowJobStatus,
    DataflowJobType,
    _DataflowJobsController,
    _fallback_to_project_id_from_variables,
    process_line_and_extract_dataflow_job_id_callback,
)
if sys.version_info < (3, 8):
    from asynctest import mock
else:
    from unittest import mock
DEFAULT_RUNNER = "DirectRunner"
BEAM_STRING = "airflow.providers.apache.beam.hooks.beam.{}"
TASK_ID = "test-dataflow-operator"
JOB_NAME = "test-dataflow-pipeline"
MOCK_UUID = UUID("cf4a56d2-8101-4217-b027-2af6216feb48")
MOCK_UUID_PREFIX = str(MOCK_UUID)[:8]
UNIQUE_JOB_NAME = f"test-dataflow-pipeline-{MOCK_UUID_PREFIX}"
TEST_TEMPLATE = "gs://dataflow-templates/wordcount/template_file"
PARAMETERS = {
    "inputFile": "gs://dataflow-samples/shakespeare/kinglear.txt",
    "output": "gs://test/output/my_output",
}
TEST_ENVIRONMENT = {}
PY_FILE = "apache_beam.examples.wordcount"
JAR_FILE = "unitest.jar"
JOB_CLASS = "com.example.UnitTest"
PY_OPTIONS = ["-m"]
DATAFLOW_VARIABLES_PY = {"project": "test", "staging_location": "gs://test/staging", "labels": {"foo": "bar"}}
DATAFLOW_VARIABLES_JAVA = {