<!--

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 -->


"""Async hook class for dataflow service."""
    sync_hook_class = DataflowHook
    def __init__(self, **kwargs):
        if kwargs.get("delegate_to") is not None:
            raise RuntimeError(
                "The `delegate_to` parameter has been deprecated before and finally removed in this version"
                " of Google Provider. You MUST convert it to `impersonate_chain`"
            )
        super().__init__(**kwargs)
    async def initialize_client(self, client_class):
        """
        Initialize object of the given class.
        Method is used to initialize asynchronous client. Because of the big amount of the classes which are
        used for Dataflow service it was decided to initialize them the same way with credentials which are
        received from the method of the GoogleBaseHook class.
        :param client_class: Class of the Google cloud SDK
        """
        credentials = (await self.get_sync_hook()).get_credentials()
        return client_class(
            credentials=credentials,
        )
    async def get_project_id(self) -> str:
        project_id = (await self.get_sync_hook()).project_id
        return project_id
    async def get_job(