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.
"Google Cloud Dataflow job name-2 was unknown state: JOB_STATE_UNKNOWN",
            ),
            (
                DataflowJobType.JOB_TYPE_BATCH,
                DataflowJobStatus.JOB_STATE_UNKNOWN,
                "Google Cloud Dataflow job name-2 was unknown state: JOB_STATE_UNKNOWN",
            ),
            (
                DataflowJobType.JOB_TYPE_BATCH,
                DataflowJobStatus.JOB_STATE_CANCELLED,
                "Google Cloud Dataflow job name-2 was cancelled\\.",
            ),
            (
                DataflowJobType.JOB_TYPE_STREAMING,
                DataflowJobStatus.JOB_STATE_CANCELLED,
                "Google Cloud Dataflow job name-2 was cancelled\\.",
            ),
            (
                DataflowJobType.JOB_TYPE_BATCH,
                DataflowJobStatus.JOB_STATE_DRAINED,
                "Google Cloud Dataflow job name-2 was drained\\.",
            ),
            (
                DataflowJobType.JOB_TYPE_STREAMING,
                DataflowJobStatus.JOB_STATE_DRAINED,
                "Google Cloud Dataflow job name-2 was drained\\.",
            ),
            (
                DataflowJobType.JOB_TYPE_BATCH,
                DataflowJobStatus.JOB_STATE_UPDATED,