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 f8b8461e819 Make SparkSubmitOperator durable execution inert below
Airflow 3.3 (#71534)
f8b8461e819 is described below
commit f8b8461e8191f88e72ad8c05b248e0385c21db99
Author: Amogh Desai <[email protected]>
AuthorDate: Tue Aug 18 10:50:14 2026 +0530
Make SparkSubmitOperator durable execution inert below Airflow 3.3 (#71534)
---
providers/apache/spark/docs/operators.rst | 15 +++++++--
.../apache/spark/operators/spark_submit.py | 37 ++++++++++++++++------
.../apache/spark/operators/test_spark_submit.py | 26 +++++++++++++--
3 files changed, 64 insertions(+), 14 deletions(-)
diff --git a/providers/apache/spark/docs/operators.rst
b/providers/apache/spark/docs/operators.rst
index e0060f6db02..6da0e87fce6 100644
--- a/providers/apache/spark/docs/operators.rst
+++ b/providers/apache/spark/docs/operators.rst
@@ -212,8 +212,12 @@ The reconnection polling calls the Spark standalone REST
API
See :doc:`connections/spark-submit` for how to configure these fields.
.. note::
- Crash recovery in cluster mode requires Airflow 3.3+ (``task_state_store``
support). On earlier
- versions the operator falls back to the previous behavior of always
submitting fresh.
+ Crash recovery in cluster mode requires Airflow 3.3+ (``task_state_store``
support). Below
+ 3.3, ``durable`` has no effect: setting it explicitly only emits a
warning, and the operator
+ always submits fresh, exactly as before this feature existed. The
deprecated
+ ``reconnect_on_retry`` parameter (the original name for this same feature,
superseded almost
+ immediately) still emits a deprecation warning on every Airflow version
and maps onto
+ ``durable``.
Clearing a task is treated the same as a retry, which matters specifically for
a task whose driver
already succeeded: clearing does not delete the stored driver ID, so the next
attempt reads it
@@ -302,6 +306,13 @@ the application is submitted:
yarn_track_via_rm_api=True,
)
+On Airflow 3.3+, YARN cluster mode with ``durable=True`` (the default) requires
+``yarn_track_via_rm_api=True`` -- the ResourceManager REST API is what makes
checking application
+status on retry possible. Without it, the operator raises a ``ValueError`` at
task start rather
+than silently falling back to a fire-and-forget submission. Below 3.3,
``durable`` has no effect
+at all, so this requirement doesn't apply there either: durable execution
isn't active to have a
+prerequisite for.
+
For Kerberized clusters, install ``requests-kerberos`` in the Airflow
environment. When the
Spark connection has both ``keytab`` and ``principal`` configured, Airflow
automatically uses
``HTTPKerberosAuth()`` for the ResourceManager REST requests.
diff --git
a/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
b/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
index 48d70155326..1a9d2748a65 100644
---
a/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
+++
b/providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py
@@ -37,21 +37,33 @@ try:
except ImportError:
kube_client = None # type: ignore[assignment]
+_DURABLE_UNSET = object()
+
+
+def _warn_and_disable_durable_pre_3_3(durable: Any) -> bool:
+ """Shared by the <3.3 compat stub: durable has no effect below 3.3, warn
if it was set."""
+ if durable is not _DURABLE_UNSET:
+ warnings.warn(
+ "`durable` has no effect on Airflow versions below 3.3.",
+ UserWarning,
+ stacklevel=3,
+ )
+ return False
+
+
try:
from airflow.sdk import ResumableJobMixin
except ImportError:
- # Airflow 2 compat.
- # ResumableJobMixin does not exist in Airflow 2, so we need to add a stub
to make it
- # behave as before
+ # ResumableJobMixin only exists on Airflow 3.3+; this provider still
targets older
+ # versions. Drop this fallback once the provider's minimum Airflow version
is >=3.3.
class ResumableJobMixin: # type: ignore[no-redef]
- """Airflow 2 stub — no task_state_store, always submits fresh."""
+ """Airflow <3.3 stub, task_state_store unavailable, always submits
fresh."""
external_id_key: str = "remote_job_id"
- def __init__(self, *, durable: bool = True, **kwargs: Any) -> None:
- # Accept durable so the kwarg doesn't leak to BaseOperator; crash
recovery is a no-op here.
+ def __init__(self, *, durable: Any = _DURABLE_UNSET, **kwargs: Any) ->
None:
super().__init__(**kwargs)
- self.durable = durable
+ self.durable = _warn_and_disable_durable_pre_3_3(durable)
def execute_resumable(self, context):
external_id = self.submit_job(context)
@@ -326,6 +338,9 @@ class SparkSubmitOperator(ResumableJobMixin, BaseOperator):
:param durable: When ``True`` (the default), the external job ID is
persisted to task state
store before polling begins so that a worker crash and retry
reconnects to the existing job
instead of submitting a fresh one. Set to ``False`` to always submit a
new job on retry.
+ Requires Airflow 3.3 or newer; below that, ``durable`` has no effect
-- setting it
+ explicitly only emits a warning.
+ :param reconnect_on_retry: deprecated, use ``durable`` instead.
"""
# Generic key used across all Spark deployment modes (standalone driver ID,
@@ -405,13 +420,15 @@ class SparkSubmitOperator(ResumableJobMixin,
BaseOperator):
) -> None:
if reconnect_on_retry is not None:
warnings.warn(
- "reconnect_on_retry is renamed to durable.",
+ "`reconnect_on_retry` is deprecated and will be removed once
this provider's "
+ "minimum supported Airflow version reaches 3.3. Use `durable`
instead.",
AirflowProviderDeprecationWarning,
stacklevel=2,
)
- kwargs.setdefault("durable", reconnect_on_retry)
+ if durable is None:
+ durable = reconnect_on_retry
# Named here (not left to **kwargs) so default_args={"durable": ...}
reaches it on every
- # supported Airflow version; applied after reconnect_on_retry so an
explicit durable wins.
+ # supported Airflow version.
if durable is not None:
kwargs["durable"] = durable
super().__init__(**kwargs)
diff --git
a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
index 3b525582ed7..c55cb10f3eb 100644
---
a/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
+++
b/providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py
@@ -28,7 +28,11 @@ import pytest
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models import DagRun, TaskInstance
from airflow.models.dag import DAG
-from airflow.providers.apache.spark.operators.spark_submit import
SparkSubmitOperator
+from airflow.providers.apache.spark.operators.spark_submit import (
+ _DURABLE_UNSET,
+ SparkSubmitOperator,
+ _warn_and_disable_durable_pre_3_3,
+)
from airflow.providers.common.compat.sdk import timezone
from airflow.utils.types import DagRunType
@@ -598,7 +602,10 @@ class TestSparkSubmitOperatorResumable:
operator = self._make_operator(reconnect_on_retry=False)
assert len(w) == 1
assert issubclass(w[0].category, AirflowProviderDeprecationWarning)
- assert "reconnect_on_retry" in str(w[0].message)
+ assert str(w[0].message) == (
+ "`reconnect_on_retry` is deprecated and will be removed once this
provider's "
+ "minimum supported Airflow version reaches 3.3. Use `durable`
instead."
+ )
assert operator.durable is False
def test_default_args_durable_reaches_operator(self):
@@ -1129,3 +1136,18 @@ class TestSparkSubmitOperatorK8sTracking:
operator.execute(context={"task_state_store": task_store})
assert task_store.get("spark_job_id") is None
+
+
+class TestWarnAndDisableDurableAirflowPre3_3:
+ def test_no_warning_when_unset(self):
+ with warnings.catch_warnings(record=True) as caught:
+ warnings.simplefilter("always")
+ result = _warn_and_disable_durable_pre_3_3(_DURABLE_UNSET)
+ assert result is False
+ assert caught == []
+
+ @pytest.mark.parametrize("value", [True, False])
+ def test_warns_and_disables_when_explicitly_set(self, value):
+ with pytest.warns(UserWarning, match="durable.*no effect"):
+ result = _warn_and_disable_durable_pre_3_3(value)
+ assert result is False
