deduplicate running jobs on BigQueryInsertJobOperator (#17496)

Author: Hossein Torabi

airflow/providers/google/cloud/operators/bigquery.py

@@ -2247,16 +2247,13 @@ def _submit_job(
         hook: BigQueryHook,
         job_id: str,
     ) -> BigQueryJob:
-        # Submit a new job
-        job = hook.insert_job(
+        # Submit a new job and wait for it to complete and get the result.
+        return hook.insert_job(
             configuration=self.configuration,
             project_id=self.project_id,
             location=self.location,
             job_id=job_id,
         )
-        # Start the job and wait for it to complete and get the result.
-        job.result()
-        return job
 
     @staticmethod
     def _handle_job_error(job: BigQueryJob) -> None:

tests/providers/google/cloud/operators/test_bigquery.py

@@ -930,7 +930,7 @@ def test_execute_reattach(self, mock_hook, mock_md5):
             }
         }
 
-        mock_hook.return_value.insert_job.return_value.result.side_effect = Conflict("any")
+        mock_hook.return_value.insert_job.side_effect = Conflict("any")
         job = MagicMock(
             job_id=real_job_id,
             error_result=False,