<!--

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


GoogleBaseAsyncHook,
    GoogleBaseHook,
)
from airflow.utils.log.logging_mixin import LoggingMixin
from airflow.utils.timeout import timeout
DEFAULT_DATAFLOW_LOCATION = "us-central1"
JOB_ID_PATTERN = re.compile(
    r"Submitted job: (?P<job_id_java>.*)|Created job with id: \[(?P<job_id_python>.*)\]"
)
T = TypeVar("T", bound=Callable)
def process_line_and_extract_dataflow_job_id_callback(
    on_new_job_id_callback: Callable[[str], None] | None
) -> Callable[[str], None]:
    """Build callback that triggers the specified function.
    The returned callback is intended to be used as ``process_line_callback`` in
    :py:class:`~airflow.providers.apache.beam.hooks.beam.BeamCommandRunner`.
    :param on_new_job_id_callback: Callback called when the job ID is known
    """
    def _process_line_and_extract_job_id(line: str) -> None:
        
        if on_new_job_id_callback is None:
            return
        matched_job = JOB_ID_PATTERN.search(line)
        if matched_job is None:
            return
        job_id = matched_job.group("job_id_java") or matched_job.group("job_id_python")
        on_new_job_id_callback(job_id)
    return _process_line_and_extract_job_id
def _fallback_variable_parameter(parameter_name: str, variable_key_name: str) -> Callable[[T], T]:
    def _wrapper(func: T) -> T: