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.
job_list_nest_method = (self.mock_dataflow
                                .projects.return_value.
                                locations.return_value.
                                jobs.return_value.list_next)
        job_list_nest_method.return_value = None
        
        dataflow_job = _DataflowJobsController(
            dataflow=self.mock_dataflow,
            project_number=TEST_PROJECT,
            name=UNIQUE_JOB_NAME,
            location=TEST_LOCATION,
            poll_sleep=10,
            job_id=TEST_JOB_ID,
            num_retries=20,
            multiple_jobs=False,
            drain_pipeline=drain_pipeline,
            cancel_timeout=None,
        )
        dataflow_job.cancel()
        get_method.assert_called_once_with(jobId=TEST_JOB_ID, location=TEST_LOCATION, projectId=TEST_PROJECT)
        get_method.return_value.execute.assert_called_once_with(num_retries=20)
        mock_update = self.mock_dataflow.projects.return_value.locations.return_value.jobs.return_value.update
        mock_update.assert_called_once_with(
            body={"requestedState": requested_state},
            jobId="test-job-id",
            location=TEST_LOCATION,
            projectId="test-project",
        )
        mock_update.return_value.execute.assert_called_once_with(num_retries=20)
    def test_dataflow_job_cancel_job_no_running_jobs(self):