Added missing project_id to the wait_for_job (#24020)

apache · May 30, 2022 · 4a52507 · 4a52507
1 parent 56fd040
commit 4a52507
Show file tree

Hide file tree

Showing 2 changed files with 4 additions and 0 deletions.
diff --git a/airflow/providers/apache/beam/operators/beam.py b/airflow/providers/apache/beam/operators/beam.py
@@ -319,6 +319,7 @@ def execute(self, context: 'Context'):
                         location=self.dataflow_config.location,
                         job_id=self.dataflow_job_id,
                         multiple_jobs=False,
+                        project_id=self.dataflow_config.project_id,
                     )
                 return {"dataflow_job_id": self.dataflow_job_id}
             else:
@@ -600,6 +601,7 @@ def execute(self, context: 'Context'):
                         location=self.dataflow_config.location,
                         job_id=self.dataflow_job_id,
                         multiple_jobs=False,
+                        project_id=self.dataflow_config.project_id,
                     )
                 return {"dataflow_job_id": self.dataflow_job_id}
             else:

diff --git a/tests/providers/apache/beam/operators/test_beam.py b/tests/providers/apache/beam/operators/test_beam.py
@@ -149,6 +149,7 @@ def test_exec_dataflow_runner(self, gcs_hook, dataflow_hook_mock, beam_hook_mock
             job_name=job_name,
             location='us-central1',
             multiple_jobs=False,
+            project_id=dataflow_config.project_id,
         )
         dataflow_hook_mock.return_value.provide_authorized_gcloud.assert_called_once_with()
 
@@ -415,6 +416,7 @@ def test_exec_dataflow_runner(self, gcs_hook, dataflow_hook_mock, beam_hook_mock
             job_name=job_name,
             location='us-central1',
             multiple_jobs=False,
+            project_id=dataflow_config.project_id,
         )
         dataflow_hook_mock.return_value.provide_authorized_gcloud.assert_called_once_with()