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, mock_callback_on_job_id, mock_dataflow_wait_for_done, mock_uuid
    ):
        mock_beam_start_java_pipeline = self.dataflow_hook.beam_hook.start_java_pipeline
        mock_uuid.return_value = MOCK_UUID
        on_new_job_id_callback = MagicMock()
        job_name = f"{JOB_NAME}-{MOCK_UUID_PREFIX}"
        passed_variables: dict[str, Any] = copy.deepcopy(DATAFLOW_VARIABLES_JAVA)
        passed_variables["mock-option"] = ["a.whl", "b.whl"]
        with pytest.warns(DeprecationWarning, match="This method is deprecated"):
            self.dataflow_hook.start_java_dataflow(
                job_name=JOB_NAME,
                variables=passed_variables,
                jar=JAR_FILE,
                job_class=JOB_CLASS,
                on_new_job_id_callback=on_new_job_id_callback,
            )
        expected_variables = copy.deepcopy(passed_variables)
        expected_variables["jobName"] = job_name
        expected_variables["region"] = DEFAULT_DATAFLOW_LOCATION
        expected_variables["labels"] = '{"foo":"bar"}'
        mock_callback_on_job_id.assert_called_once_with(on_new_job_id_callback)
        mock_beam_start_java_pipeline.assert_called_once_with(