<!--

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


#
#
#
"""This module contains a Google Dataflow Hook."""
from __future__ import annotations
import functools
import json
import re
import shlex
import subprocess
import time
import uuid
import warnings
from copy import deepcopy
from typing import Any, Callable, Generator, Sequence, TypeVar, cast
from google.cloud.dataflow_v1beta3 import GetJobRequest, Job, JobState, JobsV1Beta3AsyncClient, JobView
from googleapiclient.discovery import build
from airflow.exceptions import AirflowException, AirflowProviderDeprecationWarning
from airflow.providers.apache.beam.hooks.beam import BeamHook, BeamRunnerType, beam_options_to_args
from airflow.providers.google.common.hooks.base_google import (
    PROVIDE_PROJECT_ID,