<!--

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


"""
        Helper method to check the state of one job in dataflow for this task if job failed raise exception.
        :return: True if job is done.
        :raise: Exception
        """
        if self._wait_until_finished is None:
            wait_for_running = job.get("type") == DataflowJobType.JOB_TYPE_STREAMING
        else:
            wait_for_running = not self._wait_until_finished
        if job["currentState"] == DataflowJobStatus.JOB_STATE_DONE:
            return True
        elif job["currentState"] == DataflowJobStatus.JOB_STATE_FAILED:
            raise Exception(f"Google Cloud Dataflow job {job['name']} has failed.")
        elif job["currentState"] == DataflowJobStatus.JOB_STATE_CANCELLED:
            raise Exception(f"Google Cloud Dataflow job {job['name']} was cancelled.")
        elif job["currentState"] == DataflowJobStatus.JOB_STATE_DRAINED:
            raise Exception(f"Google Cloud Dataflow job {job['name']} was drained.")
        elif job["currentState"] == DataflowJobStatus.JOB_STATE_UPDATED:
            raise Exception(f"Google Cloud Dataflow job {job['name']} was updated.")
        elif job["currentState"] == DataflowJobStatus.JOB_STATE_RUNNING and wait_for_running:
            return True
        elif job["currentState"] in DataflowJobStatus.AWAITING_STATES:
            return self._wait_until_finished is False
        self.log.debug("Current job: %s", str(job))
        raise Exception(f"Google Cloud Dataflow job {job['name']} was unknown state: {job['currentState']}")
    def wait_for_done(self) -> None: