This is an automated email from the ASF dual-hosted git repository.
shahar1 pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new a12bf56a328 Honor cancel_on_kill in deferred GCS to BigQuery loads
(#74358)
a12bf56a328 is described below
commit a12bf56a3287055bb284cdd8bc9e5caeddac71c6
Author: Sepuri Sai Krishna <[email protected]>
AuthorDate: Sun Nov 8 20:09:34 2026 +0530
Honor cancel_on_kill in deferred GCS to BigQuery loads (#74358)
---
.../google/cloud/transfers/gcs_to_bigquery.py | 1 +
.../google/cloud/transfers/test_gcs_to_bigquery.py | 29 ++++++++++++++++++++++
2 files changed, 30 insertions(+)
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
index 5f59eff489d..4c3af87a077 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
@@ -496,6 +496,7 @@ class GCSToBigQueryOperator(BaseOperator):
project_id=self.project_id or self.hook.project_id,
location=self.location or self.hook.location,
impersonation_chain=self.impersonation_chain,
+ cancel_on_kill=self.cancel_on_kill,
),
method_name="execute_complete",
)
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
index 496207fbf40..157d7a80289 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
@@ -2324,6 +2324,35 @@ class TestAsyncGCSToBigQueryOperator:
trigger_cls =
session.scalar(select(Trigger.classpath).where(Trigger.id == ti.trigger_id))
assert trigger_cls ==
"airflow.providers.google.cloud.triggers.bigquery.BigQueryInsertJobTrigger"
+ @pytest.mark.parametrize("cancel_on_kill", [True, False])
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryInsertJobTrigger"))
+ @mock.patch(GCS_TO_BQ_PATH.format("GCSToBigQueryOperator.defer"))
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
+ def test_execute_deferrable_hands_cancel_on_kill_to_the_trigger(
+ self, hook, mock_defer, trigger, cancel_on_kill
+ ):
+ hook.return_value.insert_job.return_value =
MagicMock(job_id=REAL_JOB_ID, error_result=False)
+ hook.return_value.generate_job_id.return_value = REAL_JOB_ID
+ hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET,
TABLE)
+
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ write_disposition=WRITE_DISPOSITION,
+ schema_fields=SCHEMA_FIELDS,
+ external_table=False,
+ autodetect=True,
+ deferrable=True,
+ cancel_on_kill=cancel_on_kill,
+ project_id=JOB_PROJECT_ID,
+ )
+
+ operator.execute(context=MagicMock())
+
+ assert trigger.call_args.kwargs["cancel_on_kill"] is cancel_on_kill
+
@pytest.mark.db_test
def
test_execute_without_external_table_async_should_throw_ex_when_event_status_error(
self, create_task_instance, session
