This is an automated email from the ASF dual-hosted git repository.
Miretpl 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 274035646e1 Fix unformatted Kubernetes pod trigger error messages
(#73034)
274035646e1 is described below
commit 274035646e19083014a4e1715c408b52c13ea2c9
Author: rjgoyln <[email protected]>
AuthorDate: Sun Sep 20 01:50:02 2026 +0800
Fix unformatted Kubernetes pod trigger error messages (#73034)
Both task instance lookups built their exception the way a logging call
is written, so the placeholders were never substituted and the raised
message reached the user as a tuple repr, hiding the Dag, task and run
identifiers it was meant to carry.
Co-authored-by: Przemysław Mirowski
<[email protected]>
---
.../providers/cncf/kubernetes/triggers/pod.py | 21 +++++-------
.../unit/cncf/kubernetes/triggers/test_pod.py | 40 ++++++++++++++++++++--
2 files changed, 46 insertions(+), 15 deletions(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py
index 2b7c7f2d5ec..c72a8228ff2 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/pod.py
@@ -454,13 +454,11 @@ class KubernetesPodTrigger(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):
@@ -485,13 +483,12 @@ class KubernetesPodTrigger(BaseTrigger):
try:
return task_states_response[self.task_instance.run_id][ti_key]
except KeyError:
- raise AirflowException(
- "TaskInstance with dag_id: %s, task_id: %s, run_id: %s and
map_index: %s is not found",
- self.task_instance.dag_id,
- self.task_instance.task_id,
- self.task_instance.run_id,
- self.task_instance.map_index,
+ msg = (
+ f"TaskInstance with dag_id: {self.task_instance.dag_id}, "
+ f"task_id: {self.task_instance.task_id}, run_id:
{self.task_instance.run_id} "
+ f"and map_index: {self.task_instance.map_index} is not
found"
)
+ raise AirflowException(msg)
else:
task_instance = await sync_to_async(self.get_task_instance)() #
type: ignore[call-arg]
return task_instance.state
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py
index 9889726ab58..b27ad42eba5 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/triggers/test_pod.py
@@ -23,15 +23,18 @@ import datetime
import logging
import time
from asyncio import Future
+from types import SimpleNamespace
from unittest import mock
from unittest.mock import MagicMock
import pytest
from kubernetes.client import models as k8s
from pendulum import DateTime
+from sqlalchemy.orm.session import Session
from airflow.providers.cncf.kubernetes.triggers.pod import ContainerState,
KubernetesPodTrigger
from airflow.providers.cncf.kubernetes.utils.pod_manager import PodPhase
+from airflow.providers.common.compat.sdk import AirflowException
from airflow.triggers.base import TriggerEvent
from airflow.utils.state import TaskInstanceState
@@ -978,8 +981,6 @@ class TestKubernetesPodTrigger:
# response is missing the expected (composite) key, so callers
# like ``safe_to_cancel`` keep the same behaviour they had before
# the lookup was fixed.
- from airflow.exceptions import AirflowException
-
run_id = "manual__2026-05-21T00:00:00+00:00"
# Response has the run_id but not the (``map_group.task_a``, ``2``)
# entry -- e.g. supervisor has not observed the TI yet.
@@ -996,9 +997,42 @@ class TestKubernetesPodTrigger:
dag_id="my_dag", task_id="map_group.task_a", run_id=run_id,
map_index=2
)
- with pytest.raises(AirflowException, match="TaskInstance with dag_id"):
+ with pytest.raises(AirflowException) as exc_info:
await trigger.get_task_state()
+ assert str(exc_info.value) == (
+ "TaskInstance with dag_id: my_dag, task_id: map_group.task_a, "
+ f"run_id: {run_id} and map_index: 2 is not found"
+ )
+
+ @pytest.mark.skipif(
+ AIRFLOW_V_3_0_PLUS,
+ reason="get_task_instance reads the metadata DB on Airflow 2.x only",
+ )
+ def test_get_task_instance_reports_identifiers_when_row_missing(self):
+ run_id = "manual__2026-05-21T00:00:00+00:00"
+ session = mock.MagicMock(spec=Session)
+ session.scalar.return_value = None
+
+ trigger = KubernetesPodTrigger(
+ pod_name=POD_NAME,
+ pod_namespace=NAMESPACE,
+ base_container_name=BASE_CONTAINER_NAME,
+ trigger_start_time=TRIGGER_START_TIME,
+ schedule_timeout=STARTUP_TIMEOUT_SECS,
+ )
+ trigger.task_instance = SimpleNamespace(
+ dag_id="my_dag", task_id="my_task", run_id=run_id, map_index=-1
+ )
+
+ with pytest.raises(AirflowException) as exc_info:
+ trigger.get_task_instance(session=session)
+
+ assert str(exc_info.value) == (
+ "TaskInstance with dag_id: my_dag, task_id: my_task, "
+ f"run_id: {run_id} and map_index: -1 is not found"
+ )
+
@pytest.mark.skipif(
AIRFLOW_V_3_3_PLUS,
reason="Legacy cleanup path runs only on Airflow < 3.3",
