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 bd06decb0dc Fix unformatted Google trigger error messages (#72697)
bd06decb0dc is described below
commit bd06decb0dc59a94149864474c3625921ac49fa4
Author: rjgoyln <[email protected]>
AuthorDate: Thu Sep 17 02:50:07 2026 +0800
Fix unformatted Google trigger error messages (#72697)
---
.../providers/google/cloud/triggers/bigquery.py | 20 +++----
.../providers/google/cloud/triggers/dataproc.py | 63 +++++++++-------------
.../unit/google/cloud/triggers/test_dataproc.py | 47 +++++++++++++++-
3 files changed, 80 insertions(+), 50 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
b/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
index e11059dfbb8..244043ca5d9 100644
--- a/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
+++ b/providers/google/src/airflow/providers/google/cloud/triggers/bigquery.py
@@ -132,13 +132,11 @@ class BigQueryInsertJobTrigger(BaseTrigger):
)
)
if task_instance is None:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_instance
async def get_task_state(self):
@@ -156,13 +154,11 @@ class BigQueryInsertJobTrigger(BaseTrigger):
try:
task_state = task_states_response[ti.run_id][ti.task_id]
except Exception:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_state
async def safe_to_cancel(self) -> bool:
diff --git
a/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
b/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
index 17eba607d58..08adba0fbee 100644
--- a/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
+++ b/providers/google/src/airflow/providers/google/cloud/triggers/dataproc.py
@@ -153,13 +153,11 @@ class DataprocSubmitTrigger(DataprocBaseTrigger):
)
)
if task_instance is None:
- raise AirflowException(
- "TaskInstance with dag_id: %s,task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_instance
async def get_task_state(self):
@@ -177,13 +175,11 @@ class DataprocSubmitTrigger(DataprocBaseTrigger):
try:
task_state = task_states_response[ti.run_id][ti.task_id]
except Exception:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_state
async def safe_to_cancel(self) -> bool:
@@ -311,13 +307,11 @@ class DataprocSubmitJobDirectTrigger(DataprocBaseTrigger):
)
)
if task_instance is None:
- raise RuntimeError(
- "TaskInstance with dag_id: %s,task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise RuntimeError(msg)
return task_instance
async def get_task_state(self):
@@ -335,13 +329,11 @@ class DataprocSubmitJobDirectTrigger(DataprocBaseTrigger):
try:
task_state = task_states_response[ti.run_id][ti.task_id]
except Exception:
- raise RuntimeError(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise RuntimeError(msg)
return task_state
async def safe_to_cancel(self) -> bool:
@@ -456,13 +448,11 @@ class DataprocClusterTrigger(DataprocBaseTrigger):
)
)
if task_instance is None:
- raise AirflowException(
- "TaskInstance with dag_id: %s,task_id: %s, run_id: %s and
map_index: %s is not found.",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_instance
async def get_task_state(self):
@@ -480,13 +470,11 @@ class DataprocClusterTrigger(DataprocBaseTrigger):
try:
task_state = task_states_response[ti.run_id][ti.task_id]
except Exception:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- ti.dag_id,
- ti.task_id,
- ti.run_id,
- ti.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {ti.dag_id}, task_id:
{ti.task_id}, "
+ f"run_id: {ti.run_id} and map_index: {ti.map_index} is not
found"
)
+ raise AirflowException(msg)
return task_state
async def safe_to_cancel(self) -> bool:
@@ -550,7 +538,8 @@ class DataprocClusterTrigger(DataprocBaseTrigger):
self.log.info("Deleted cluster %s during cancellation.",
self.cluster_name)
except Exception as e:
self.log.error("Error during cancellation handling: %s", e)
- raise AirflowException("Error during cancellation handling:
%s", e)
+ msg = f"Error during cancellation handling: {e}"
+ raise AirflowException(msg)
async def fetch_cluster(self) -> Cluster:
"""Fetch the cluster status."""
diff --git a/providers/google/tests/unit/google/cloud/triggers/test_dataproc.py
b/providers/google/tests/unit/google/cloud/triggers/test_dataproc.py
index 8e966830215..ccc7a9f91ad 100644
--- a/providers/google/tests/unit/google/cloud/triggers/test_dataproc.py
+++ b/providers/google/tests/unit/google/cloud/triggers/test_dataproc.py
@@ -20,6 +20,7 @@ import asyncio
import contextlib
import logging
from asyncio import CancelledError, Future, sleep
+from types import SimpleNamespace
from unittest import mock
import pytest
@@ -27,6 +28,7 @@ from google.cloud.dataproc_v1 import Batch, Cluster,
ClusterStatus, Job, JobStat
from google.protobuf.any_pb2 import Any
from google.rpc.status_pb2 import Status
+from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.google.cloud.triggers.dataproc import (
DataprocBatchTrigger,
DataprocClusterTrigger,
@@ -37,7 +39,7 @@ from airflow.providers.google.cloud.triggers.dataproc import (
from airflow.providers.google.cloud.utils.dataproc import DataprocOperationType
from airflow.triggers.base import TriggerEvent
-from tests_common.test_utils.version_compat import AIRFLOW_V_3_3_PLUS
+from tests_common.test_utils.version_compat import AIRFLOW_V_3_0_PLUS,
AIRFLOW_V_3_3_PLUS
TEST_PROJECT_ID = "project-id"
TEST_REGION = "region"
@@ -378,6 +380,49 @@ class TestDataprocClusterTrigger:
assert mock_delete_cluster.call_count == 0
mock_delete_cluster.assert_not_called()
+ @pytest.mark.asyncio
+ @pytest.mark.skipif(
+ not AIRFLOW_V_3_0_PLUS, reason="The task state is read through the
task SDK on Airflow 3 only"
+ )
+ @pytest.mark.parametrize("map_index", [-1, 3])
+
@mock.patch("airflow.sdk.execution_time.task_runner.RuntimeTaskInstance.get_task_states",
autospec=True)
+ async def test_get_task_state_missing_task_instance_reports_identifiers(
+ self, mock_get_task_states, cluster_trigger, map_index
+ ):
+ cluster_trigger.task_instance = SimpleNamespace(
+ dag_id="test_dag", task_id="test_task", run_id="test_run",
map_index=map_index
+ )
+ mock_get_task_states.return_value = {}
+
+ with pytest.raises(AirflowException) as exc_info:
+ await cluster_trigger.get_task_state()
+
+ assert str(exc_info.value) == (
+ "TaskInstance with dag_id: test_dag, task_id: test_task, "
+ f"run_id: test_run and map_index: {map_index} is not found"
+ )
+
+ @pytest.mark.asyncio
+
@mock.patch("airflow.providers.google.cloud.triggers.dataproc.DataprocClusterTrigger.get_async_hook")
+
@mock.patch("airflow.providers.google.cloud.triggers.dataproc.DataprocClusterTrigger.get_sync_hook")
+
@mock.patch("airflow.providers.google.cloud.triggers.dataproc.DataprocClusterTrigger.safe_to_cancel")
+ async def test_cluster_trigger_cancellation_failure_reports_cause(
+ self, mock_safe_to_cancel, mock_get_sync_hook, mock_get_async_hook,
cluster_trigger
+ ):
+ mock_safe_to_cancel.return_value = True
+ mock_get_async_hook.return_value.get_cluster.return_value = Future()
+ mock_get_sync_hook.return_value.delete_cluster.side_effect =
ValueError("cluster is already deleted")
+
+ async_gen = cluster_trigger.run()
+ task = asyncio.create_task(async_gen.__anext__())
+ await sleep(0)
+ task.cancel()
+
+ with pytest.raises(AirflowException) as exc_info:
+ await task
+
+ assert str(exc_info.value) == "Error during cancellation handling:
cluster is already deleted"
+
class TestDataprocBatchTrigger:
def
test_async_create_batch_trigger_serialization_should_execute_successfully(self,
batch_trigger):
