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.
@_fallback_to_project_id_from_variables
            def test_fn(self, *args, **kwargs):
                mock_instance(*args, **kwargs)
        FixtureFallback().test_fn(project_id="TEST")
        mock_instance.assert_called_once_with(project_id="TEST")
    def test_support_project_id_from_variable_parameter(self):
        mock_instance = mock.MagicMock()
        class FixtureFallback:
            @_fallback_to_project_id_from_variables
            def test_fn(self, *args, **kwargs):
                mock_instance(*args, **kwargs)
        FixtureFallback().test_fn(variables={"project": "TEST"})
        mock_instance.assert_called_once_with(project_id="TEST", variables={})
    def test_raise_exception_on_conflict(self):
        mock_instance = mock.MagicMock()
        class FixtureFallback:
            @_fallback_to_project_id_from_variables
            def test_fn(self, *args, **kwargs):
                mock_instance(*args, **kwargs)
        with pytest.raises(
            AirflowException,
            match="The mutually exclusive parameter `project_id` and `project` key in `variables` parameter "
            "are both present\\. Please remove one\\.",
        ):
            FixtureFallback().test_fn(variables={"project": "TEST"}, project_id="TEST2")
    def test_raise_exception_on_positional_argument(self):
        mock_instance = mock.MagicMock()
        class FixtureFallback: