As discussed in  https://github.com/apache/airflow/pull/15534#discussion_r620500075, we could possibly rewrite Beam Hook to remove the need of storing state in it.
echos = ";".join(f"echo {shlex.quote(line)}" for line in log.split("\n"))
        cmd = ["bash", "-c", echos]
        found_job_id = None
        def callback(job_id):
            nonlocal found_job_id
            found_job_id = job_id
        BeamCommandRunner(
            cmd, process_line_callback=process_line_and_extract_dataflow_job_id_callback(callback)
        ).wait_for_done()
        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
        BeamCommandRunner(
            cmd, process_line_callback=process_line_and_extract_dataflow_job_id_callback(callback)
        ).wait_for_done()
        assert found_job_id is None
    @mock.patch("airflow.providers.apache.beam.hooks.beam.BeamCommandRunner.log")
    @mock.patch("subprocess.Popen")
    @mock.patch("select.select")
    def test_dataflow_wait_for_done_logging(self, mock_select, mock_popen, mock_logging):