diff --git a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py b/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
index d144141b0e676..952f52612e61f 100644
--- a/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
+++ b/providers/google/src/airflow/providers/google/cloud/operators/bigquery.py
@@ -2865,6 +2865,7 @@ def execute(self, context: Any):
                 )
             self.log.info("Current state of job %s is %s", job.job_id, job.state)
             self._handle_job_error(job)
+            return self.job_id
 
     def execute_complete(self, context: Context, event: dict[str, Any]) -> str | None:
         """
diff --git a/providers/google/tests/provider_tests/google/cloud/operators/test_bigquery.py b/providers/google/tests/provider_tests/google/cloud/operators/test_bigquery.py
index 1fc8ab93cdcf9..d95872b6c39a9 100644
--- a/providers/google/tests/provider_tests/google/cloud/operators/test_bigquery.py
+++ b/providers/google/tests/provider_tests/google/cloud/operators/test_bigquery.py
@@ -1488,9 +1488,11 @@ def test_bigquery_insert_job_operator_async_finish_before_deferred(self, mock_ho
             deferrable=True,
         )
 
-        op.execute(MagicMock())
+        result = op.execute(context=MagicMock())
+
         assert not mock_defer.called
         assert "Current state of job" in caplog.text
+        assert result == real_job_id
 
     @mock.patch("airflow.providers.google.cloud.operators.bigquery.BigQueryInsertJobOperator.defer")
     @mock.patch("airflow.providers.google.cloud.operators.bigquery.BigQueryHook")