<!--

Welcome to Apache Airflow!  For a smooth issue process, try to answer the following questions.
Don't worry if they're not all applicable; just try to include what you can :-)

If you need to include code snippets or logs, please put them in fenced code
blocks.  If they're super-long, please use the details tag like
<details><summary>super-long log</summary> lots of stuff </details>

Please delete these comment blocks before submitting the issue.

-->

**Description**

Related to the discussion here: https://github.com/apache/airflow/pull/8553#discussion_r416136954

Allow user to define expected terminal state. e.g. `JOB_STATE_CANCELLED` could be not treated as failure but expected terminal state.


<!-- A short description of your feature -->


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:
            @_fallback_to_project_id_from_variables
            def test_fn(self, *args, **kwargs):
                mock_instance(*args, **kwargs)
        with pytest.raises(
            AirflowException, match="You must use keyword arguments in this methods rather than positional"
        ):
            FixtureFallback().test_fn({"project": "TEST"}, "TEST2")
class TestDataflowHook:
    def test_delegate_to_runtime_error(self):
        with pytest.raises(RuntimeError):
            DataflowHook(gcp_conn_id="GCP_CONN_ID", delegate_to="delegate_to")
    def setup_method(self):