This is an automated email from the ASF dual-hosted git repository.
amoghrajesh 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 f7360776e44 Mark KubernetesPodOperator and AgentOperator as durable
capable (#70289)
f7360776e44 is described below
commit f7360776e4450b2a58603b3cc2ebc5961b2c2bf0
Author: Amogh Desai <[email protected]>
AuthorDate: Fri Jul 24 13:28:32 2026 +0530
Mark KubernetesPodOperator and AgentOperator as durable capable (#70289)
---
.../src/airflow/providers/cncf/kubernetes/operators/pod.py | 6 +++++-
.../kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py | 3 +++
.../common/ai/src/airflow/providers/common/ai/operators/agent.py | 4 ++++
providers/common/ai/tests/unit/common/ai/operators/test_agent.py | 3 +++
4 files changed, 15 insertions(+), 1 deletion(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
index 178f61cda9a..afbe07b827a 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py
@@ -34,7 +34,7 @@ from collections.abc import Callable, Container, Iterable,
Mapping, Sequence
from contextlib import AbstractContextManager, suppress
from enum import Enum
from functools import cached_property
-from typing import TYPE_CHECKING, Any, Literal
+from typing import TYPE_CHECKING, Any, ClassVar, Literal
import kubernetes
import pendulum
@@ -283,6 +283,10 @@ class KubernetesPodOperator(BaseOperator):
# !!! Changes in KubernetesPodOperator's arguments should be also
reflected in !!!
# - airflow-core/src/airflow/decorators/__init__.pyi (by a separate PR)
+ # This operator supports durable execution directly, without
ResumableJobMixin --
+ # it reconnects via task_state_store on retry instead of resubmitting.
+ __supports_durable_execution: ClassVar[bool] = True
+
# This field can be overloaded at the instance level via
base_container_name
BASE_CONTAINER_NAME = "base"
ISTIO_CONTAINER_NAME = "istio-proxy"
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
index ea5ed5a207c..0e903ec644b 100644
--- a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
+++ b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_pod.py
@@ -2776,6 +2776,9 @@ class TestKubernetesPodOperatorDurableExecution:
assert k.durable is False
assert k.reattach_on_restart is False
+ def test_supports_durable_execution_marker(self):
+ assert
KubernetesPodOperator._KubernetesPodOperator__supports_durable_execution is True
+
class TestSuppress:
def test__suppress(self, caplog):
diff --git
a/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py
b/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py
index 5afbe3909b5..83a03bfda02 100644
--- a/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py
+++ b/providers/common/ai/src/airflow/providers/common/ai/operators/agent.py
@@ -221,6 +221,10 @@ class AgentOperator(BaseOperator, HITLReviewMixin):
deserialization_allowed_class_fields: ClassVar[tuple[str, ...]] =
("output_type",)
+ # This operator supports durable execution directly, without
ResumableJobMixin --
+ # it caches step results via task_state_store for replay on retry.
+ __supports_durable_execution: ClassVar[bool] = True
+
template_fields: Sequence[str] = (
"prompt",
"llm_conn_id",
diff --git a/providers/common/ai/tests/unit/common/ai/operators/test_agent.py
b/providers/common/ai/tests/unit/common/ai/operators/test_agent.py
index 085f8512df8..760bacb8cc7 100644
--- a/providers/common/ai/tests/unit/common/ai/operators/test_agent.py
+++ b/providers/common/ai/tests/unit/common/ai/operators/test_agent.py
@@ -759,6 +759,9 @@ class TestAgentOperatorDurable:
storage.cleanup.assert_not_called()
+ def test_supports_durable_execution_marker(self):
+ assert AgentOperator._AgentOperator__supports_durable_execution is True
+
@pytest.mark.skipif(
not AIRFLOW_V_3_1_PLUS, reason="Human in the loop is only compatible with
Airflow >= 3.1.0"
