Make SparkSubmitOperator durable execution inert below Airflow 3.3

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.

providers/apache/spark/src/airflow/providers/apache/spark/operators/spark_submit.py

@@ -37,21 +37,33 @@
 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 @@ def __init__(
     ) -> 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)

providers/apache/spark/tests/unit/apache/spark/operators/test_spark_submit.py

@@ -28,7 +28,11 @@
 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 @@ def test_reconnect_on_retry_deprecated_alias(self):
             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 @@ def test_k8s_execute_durable_false_does_not_persist_pod_id(self):
         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