<!--

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


echos = ";".join(f"echo {shlex.quote(line)}" for line in log.splitlines())
        cmd = ["bash", "-c", echos]
        found_job_id = None
        def callback(job_id):
            nonlocal found_job_id
            found_job_id = job_id
        mock_log = MagicMock()
        run_beam_command(
            cmd=cmd,
            process_line_callback=process_line_and_extract_dataflow_job_id_callback(callback),
            log=mock_log,
        )
        assert found_job_id == TEST_JOB_ID
    def test_data_flow_missing_job_id(self):
        cmd = ["echo", "unit testing"]
        found_job_id = None
        def callback(job_id):
            nonlocal found_job_id
            found_job_id = job_id
        log = MagicMock()
        run_beam_command(
            cmd=cmd,
            process_line_callback=process_line_and_extract_dataflow_job_id_callback(callback),
            log=log,
        )
        assert found_job_id is None
    @mock.patch("subprocess.Popen")
    @mock.patch("select.select")
    def test_dataflow_wait_for_done_logging(self, mock_select, mock_popen):