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.
)
        expected_variables = copy.deepcopy(DATAFLOW_VARIABLES_JAVA)
        expected_variables["jobName"] = job_name
        expected_variables["region"] = TEST_LOCATION
        expected_variables["labels"] = '{"foo":"bar"}'
        mock_callback_on_job_id.assert_called_once_with(on_new_job_id_callback)
        mock_beam_start_java_pipeline.assert_called_once_with(
            variables=expected_variables,
            jar=JAR_FILE,
            job_class=JOB_CLASS,
            process_line_callback=mock_callback_on_job_id.return_value,
        )
        mock_dataflow_wait_for_done.assert_called_once_with(
            job_id=mock.ANY, job_name=job_name, location=TEST_LOCATION, multiple_jobs=False
        )
    @pytest.mark.parametrize(
        "expected_result, job_name, append_job_name",
        [
            (JOB_NAME, JOB_NAME, False),
            ("test-example", "test_example", False),
            (f"test-dataflow-pipeline-{MOCK_UUID_PREFIX}", JOB_NAME, True),
            (f"test-example-{MOCK_UUID_PREFIX}", "test_example", True),
            ("df-job-1", "df-job-1", False),
            ("df-job", "df-job", False),
            ("dfjob", "dfjob", False),
            ("dfjob1", "dfjob1", False),
        ],
    )
    @mock.patch(DATAFLOW_STRING.format("uuid.uuid4"), return_value=MOCK_UUID)
    def test_valid_dataflow_job_name(self, _, expected_result, job_name, append_job_name):