This is an automated email from the ASF dual-hosted git repository.
shahar1 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 4d8eae3f28d Fix SparkKubernetesOperator with deferrable=True falling
through to KubernetesPodOperator.execute() creating a spurious pod (#66448)
4d8eae3f28d is described below
commit 4d8eae3f28d981bc7afca0143e3d8d24e71d89b7
Author: Pranay Kumar Karvi <[email protected]>
AuthorDate: Wed May 6 09:48:04 2026 +0530
Fix SparkKubernetesOperator with deferrable=True falling through to
KubernetesPodOperator.execute() creating a spurious pod (#66448)
---
.../cncf/kubernetes/operators/spark_kubernetes.py | 1 +
.../kubernetes/operators/test_spark_kubernetes.py | 62 ++++++++++++++++++++++
2 files changed, 63 insertions(+)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
index e1cbe0595f2..d10cee73e5c 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py
@@ -356,6 +356,7 @@ class SparkKubernetesOperator(KubernetesPodOperator):
if self.deferrable:
self.execute_async(context)
+ return
return super().execute(context)
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_spark_kubernetes.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_spark_kubernetes.py
index b5f278ad49d..9a2f739266d 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_spark_kubernetes.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/operators/test_spark_kubernetes.py
@@ -1242,6 +1242,68 @@ class TestSparkKubernetesOperator:
assert returned_pod is valid_mock_pod
+ def test_execute_deferrable_does_not_call_super(
+ self,
+ mock_is_in_cluster,
+ mock_parent_execute,
+ mock_create_namespaced_crd,
+ mock_get_namespaced_custom_object_status,
+ mock_cleanup,
+ mock_create_job_name,
+ mock_get_kube_client,
+ mock_create_pod,
+ mock_await_pod_completion,
+ mock_fetch_requested_container_logs,
+ data_file,
+ mocker,
+ ):
+ job_spec =
yaml.safe_load(data_file("spark/application_template.yaml").read_text())
+ mock_create_job_name.return_value = "test_deferrable"
+ op = SparkKubernetesOperator(
+ task_id="test_deferrable",
+ template_spec=job_spec,
+ kubernetes_conn_id="kubernetes_default_kube_config",
+ deferrable=True,
+ reattach_on_restart=False,
+ )
+ context = create_context(op)
+ mock_execute_async = mocker.patch.object(op, "execute_async")
+
+ result = op.execute(context)
+
+ mock_execute_async.assert_called_once_with(context)
+ mock_parent_execute.assert_not_called()
+ assert result is None
+
+ def test_execute_non_deferrable_calls_super(
+ self,
+ mock_is_in_cluster,
+ mock_parent_execute,
+ mock_create_namespaced_crd,
+ mock_get_namespaced_custom_object_status,
+ mock_cleanup,
+ mock_create_job_name,
+ mock_get_kube_client,
+ mock_create_pod,
+ mock_await_pod_completion,
+ mock_fetch_requested_container_logs,
+ data_file,
+ ):
+ job_spec =
yaml.safe_load(data_file("spark/application_template.yaml").read_text())
+ mock_create_job_name.return_value = "test_non_deferrable"
+ op = SparkKubernetesOperator(
+ task_id="test_non_deferrable",
+ template_spec=job_spec,
+ kubernetes_conn_id="kubernetes_default_kube_config",
+ deferrable=False,
+ reattach_on_restart=False,
+ )
+ context = create_context(op)
+
+ op.execute(context)
+
+ mock_parent_execute.assert_called_once_with(context)
+
@pytest.mark.asyncio
def test_execute_deferrable(
self,
