Fix location requirement in DataflowTemplatedJobStartOperator

VladaZakharova · Jan 23, 2024 · 96670b4 · 96670b4
1 parent 13b0930
commit 96670b4
Showing 1 changed file with 4 additions and 1 deletion.
diff --git a/airflow/providers/google/cloud/operators/dataflow.py b/airflow/providers/google/cloud/operators/dataflow.py
@@ -677,6 +677,9 @@ def set_current_job(current_job):
         options = self.dataflow_default_options
         options.update(self.options)
 
+        if not self.location:
+            self.location = DEFAULT_DATAFLOW_LOCATION
+
         self.job = self.hook.start_template_dataflow(
             job_name=self.job_name,
             variables=options,
@@ -704,7 +707,7 @@ def set_current_job(current_job):
             trigger=TemplateJobStartTrigger(
                 project_id=self.project_id,
                 job_id=job_id,
-                location=self.location if self.location else DEFAULT_DATAFLOW_LOCATION,
+                location=self.location,
                 gcp_conn_id=self.gcp_conn_id,
                 poll_sleep=self.poll_sleep,
                 impersonation_chain=self.impersonation_chain,