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.
#
#
#
from __future__ import annotations
import copy
import re
import shlex
import sys
from asyncio import Future
from typing import Any
from unittest.mock import MagicMock
from uuid import UUID
import pytest
from google.cloud.dataflow_v1beta3 import GetJobRequest, JobView
from airflow.exceptions import AirflowException
from airflow.providers.apache.beam.hooks.beam import BeamCommandRunner, BeamHook
from airflow.providers.google.cloud.hooks.dataflow import (
    DEFAULT_DATAFLOW_LOCATION,
    AsyncDataflowHook,