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 168d46aaac8 Make KPO durable execution inert below Airflow 3.3 (#71492)
168d46aaac8 is described below
commit 168d46aaac8b54b2ccb8bb8fa3a46da43b6ad84e
Author: Amogh Desai <[email protected]>
AuthorDate: Fri Aug 14 11:21:16 2026 +0530
Make KPO durable execution inert below Airflow 3.3 (#71492)
---
providers/cncf/kubernetes/docs/operators.rst | 16 +++--
.../providers/cncf/kubernetes/operators/pod.py | 49 +++++++++----
.../unit/cncf/kubernetes/operators/test_pod.py | 80 ++++++++++++++++++++--
3 files changed, 119 insertions(+), 26 deletions(-)
diff --git a/providers/cncf/kubernetes/docs/operators.rst
b/providers/cncf/kubernetes/docs/operators.rst
index 05da87f11dd..a7d1c242b12 100644
--- a/providers/cncf/kubernetes/docs/operators.rst
+++ b/providers/cncf/kubernetes/docs/operators.rst
@@ -216,9 +216,11 @@ To always create a fresh pod on retry rather than
reattaching, set ``durable=Fal
durable=False,
)
-Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. On
-earlier Airflow versions ``durable=True`` (the default) falls back to the same
label-search
-reattach behavior this operator has always used.
+Durable execution requires Airflow 3.3 or newer, since it relies on the task
state store. Below
+3.3, ``durable`` has no effect at all: setting it explicitly only emits a
warning, and its value is
+ignored either way. The deprecated ``reattach_on_restart`` parameter (default
``True``) is the
+only lever there, and it falls back to the same label-search reattach behavior
this operator has
+always used -- unchanged from before this feature existed.
The pod identity persisted in task state store isn't deleted automatically,
that only happens
when someone runs ``airflow state-store clean``. If a task's ``retry_delay``
is longer than
@@ -229,9 +231,11 @@ search can often still find the same pod, but it loses the
unambiguous reconnect
exposure to ``FoundMoreThanOnePodFailure`` if a genuine duplicate pod exists
by then. Avoid
running cleanup on a schedule shorter than your longest ``retry_delay``.
-``durable`` supersedes the deprecated ``reattach_on_restart`` parameter --
passing
-``reattach_on_restart`` still works but emits
``AirflowProviderDeprecationWarning`` (on Airflow
-3.3+) and maps its value onto ``durable``.
+``durable`` supersedes the deprecated ``reattach_on_restart`` parameter on
Airflow 3.3+, where
+passing ``reattach_on_restart`` still works and maps its value onto
``durable``. Below 3.3,
+``reattach_on_restart`` remains the only working option, since ``durable`` is
a no-op there.
+Either way, passing it emits an ``AirflowProviderDeprecationWarning``, since
the parameter will be
+removed once this provider's minimum supported Airflow version reaches 3.3.
How does XCom work?
^^^^^^^^^^^^^^^^^^^
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 f52333d605c..5a9df5645aa 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
@@ -176,16 +176,17 @@ class KubernetesPodOperator(BaseOperator):
:param in_cluster: run kubernetes client with in_cluster configuration.
:param cluster_context: context that points to kubernetes cluster.
Ignored when in_cluster is True. If None, current-context is used.
(templated)
- :param reattach_on_restart: deprecated for Airflow 3.3+, use ``durable``
instead. If the worker dies
- while the pod is running, reattach and monitor during the next try. If
False, always create a
- new pod for each try.
+ :param reattach_on_restart: deprecated, use ``durable`` instead. If the
worker dies while the pod
+ is running, reattach and monitor during the next try. If False, always
create a new pod for
+ each try. Below Airflow 3.3, this remains the only way to control that
behavior, since
+ ``durable`` has no effect there.
:param durable: if the worker dies while the pod is running, reattach and
monitor during the next
try instead of creating a duplicate pod. If False, always create a new
pod for each try.
- Supersedes ``reattach_on_restart``; on Airflow 3.3+ the reconnection
uses a persisted pod
- identity in task state store instead of a label search, removing the
ambiguity failure a label
- search can hit when more than one matching pod exists. Defaults to
``True``. On Airflow
- versions below 3.3, ``durable`` still works but falls back to the same
label-search reattach
- behavior as ``reattach_on_restart``, since task state store is
unavailable.
+ Supersedes ``reattach_on_restart`` on Airflow 3.3+, where the
reconnection uses a persisted
+ pod identity in task state store instead of a label search. Defaults
to ``True`` on Airflow
+ 3.3+. Below 3.3, ``durable`` has no effect -- setting it explicitly
only emits a warning --
+ and ``reattach_on_restart`` (default ``True``) is the only lever,
using the same
+ label-search reattach mechanism it always has.
:param labels: labels to apply to the Pod. (templated)
:param startup_timeout_seconds: timeout in seconds to startup the pod
after pod was scheduled.
:param startup_check_interval_seconds: interval in seconds to check if the
pod has already started
@@ -340,7 +341,7 @@ class KubernetesPodOperator(BaseOperator):
cluster_context: str | None = None,
labels: dict | None = None,
reattach_on_restart: bool | None = None,
- durable: bool = True,
+ durable: bool | None = None,
startup_timeout_seconds: int = 120,
startup_check_interval_seconds: int = 5,
schedule_timeout_seconds: int | None = None,
@@ -401,14 +402,34 @@ class KubernetesPodOperator(BaseOperator):
) -> None:
if reattach_on_restart is not None:
# Kept as a real named parameter (not **kwargs) so `default_args`
still applies correctly.
- if AIRFLOW_V_3_3_PLUS:
+ warnings.warn(
+ "`reattach_on_restart` is deprecated and will be removed once
this provider's "
+ "minimum supported Airflow version reaches 3.3. "
+ + (
+ "Use `durable` instead."
+ if AIRFLOW_V_3_3_PLUS
+ else "On Airflow 3.3+, use `durable` instead."
+ ),
+ AirflowProviderDeprecationWarning,
+ stacklevel=2,
+ )
+ if AIRFLOW_V_3_3_PLUS:
+ if durable is None:
+ # durable takes precedence when set, even against a conflicting
+ # reattach_on_restart; reattach_on_restart only fills the gap
when durable itself
+ # was never touched.
+ durable = True if reattach_on_restart is None else
reattach_on_restart
+ else:
+ if durable is not None:
+ # durable itself has no effect below 3.3 -- there's no task
state store to persist
+ # a pod identity to, so only reattach_on_restart's
label-search fallback is real
+ # there.
warnings.warn(
- "`reattach_on_restart` is deprecated and will be removed
in a future release. "
- "Use `durable` instead.",
- AirflowProviderDeprecationWarning,
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
stacklevel=2,
)
- durable = reattach_on_restart
+ durable = True if reattach_on_restart is None else
reattach_on_restart
super().__init__(**kwargs)
self.kubernetes_conn_id = kubernetes_conn_id
self.do_xcom_push = do_xcom_push
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 9d070272136..d259d87c623 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
@@ -2566,6 +2566,16 @@ class TestKubernetesPodOperator:
assert result.metadata.name == pod_2.metadata.name
+_REATTACH_DEPRECATION_MESSAGE_PREFIX = (
+ "`reattach_on_restart` is deprecated and will be removed once this
provider's "
+ "minimum supported Airflow version reaches 3.3. "
+)
+REATTACH_DEPRECATION_MESSAGE_PRE_3_3 = (
+ _REATTACH_DEPRECATION_MESSAGE_PREFIX + "On Airflow 3.3+, use `durable`
instead."
+)
+REATTACH_DEPRECATION_MESSAGE_3_3_PLUS = _REATTACH_DEPRECATION_MESSAGE_PREFIX +
"Use `durable` instead."
+
+
@pytest.mark.skipif(
not AIRFLOW_V_3_3_PLUS, reason="durable execution (task_state_store)
requires Airflow 3.3+"
)
@@ -2581,6 +2591,12 @@ class TestKubernetesPodOperatorDurableExecution:
patch.stopall()
+ def test_warning_message_recommends_durable_directly_on_3_3_plus(self):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_3_3_PLUS)}$"
+ ):
+ KubernetesPodOperator(task_id="task", reattach_on_restart=True)
+
def test_durable_fresh_submit_persists_pod_identity(self):
k = KubernetesPodOperator(
image="ubuntu:16.04",
@@ -2779,7 +2795,9 @@ class TestKubernetesPodOperatorDurableExecution:
@pytest.mark.parametrize("reattach_value", [True, False])
def test_reattach_on_restart_deprecation_maps_to_durable(self,
reattach_value):
- with pytest.warns(AirflowProviderDeprecationWarning,
match="reattach_on_restart"):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_3_3_PLUS)}$"
+ ):
k = KubernetesPodOperator(
task_id="task",
reattach_on_restart=reattach_value,
@@ -2787,18 +2805,22 @@ class TestKubernetesPodOperatorDurableExecution:
assert k.durable is reattach_value
assert k.reattach_on_restart is reattach_value
- def
test_reattach_on_restart_and_durable_conflict_reattach_on_restart_wins(self):
- with pytest.warns(AirflowProviderDeprecationWarning,
match="reattach_on_restart"):
+ def test_durable_wins_over_conflicting_reattach_on_restart(self):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_3_3_PLUS)}$"
+ ):
k = KubernetesPodOperator(
task_id="task",
durable=False,
reattach_on_restart=True,
)
- assert k.durable is True
- assert k.reattach_on_restart is True
+ assert k.durable is False
+ assert k.reattach_on_restart is False
def test_reattach_on_restart_via_default_args_reaches_durable(self,
dag_maker):
- with pytest.warns(AirflowProviderDeprecationWarning,
match="reattach_on_restart"):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_3_3_PLUS)}$"
+ ):
with dag_maker(dag_id="test_reattach_default_args",
default_args={"reattach_on_restart": False}):
k = KubernetesPodOperator(task_id="task")
assert k.durable is False
@@ -2808,6 +2830,52 @@ class TestKubernetesPodOperatorDurableExecution:
assert
KubernetesPodOperator._KubernetesPodOperator__supports_durable_execution is True
+class TestKubernetesPodOperatorDurableBelow3_3:
+ @pytest.mark.parametrize("durable_value", [True, False])
+ def test_durable_has_no_effect(self, durable_value):
+ with
mock.patch("airflow.providers.cncf.kubernetes.operators.pod.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ UserWarning, match=r"^`durable` has no effect on Airflow
versions below 3\.3\.$"
+ ):
+ k = KubernetesPodOperator(task_id="task",
durable=durable_value)
+ # old default (reattach_on_restart's default was True), untouched by
the ignored durable value.
+ assert k.durable is True
+ assert k.reattach_on_restart is True
+
+ @pytest.mark.parametrize("reattach_value", [True, False])
+ def test_reattach_on_restart_still_works(self, reattach_value):
+ with
mock.patch("airflow.providers.cncf.kubernetes.operators.pod.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
+ match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_PRE_3_3)}$",
+ ):
+ k = KubernetesPodOperator(task_id="task",
reattach_on_restart=reattach_value)
+ assert k.reattach_on_restart is reattach_value
+
+ def test_reattach_on_restart_wins_over_durable(self):
+ with
mock.patch("airflow.providers.cncf.kubernetes.operators.pod.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
+ match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_PRE_3_3)}$",
+ ):
+ k = KubernetesPodOperator(task_id="task", durable=False,
reattach_on_restart=True)
+ assert k.reattach_on_restart is True
+
+ def test_default_rettach_on_restart_is_true(self):
+ with
mock.patch("airflow.providers.cncf.kubernetes.operators.pod.AIRFLOW_V_3_3_PLUS",
False):
+ k = KubernetesPodOperator(task_id="task")
+ assert k.durable is True
+ assert k.reattach_on_restart is True
+
+ def test_warns_on_every_supported_airflow_version(self):
+ with
mock.patch("airflow.providers.cncf.kubernetes.operators.pod.AIRFLOW_V_3_3_PLUS",
False):
+ with pytest.warns(
+ AirflowProviderDeprecationWarning,
+ match=f"^{re.escape(REATTACH_DEPRECATION_MESSAGE_PRE_3_3)}$",
+ ):
+ KubernetesPodOperator(task_id="task", reattach_on_restart=True)
+
+
class TestSuppress:
def test__suppress(self, caplog):
with _optionally_suppress(ValueError):
