Fix deferrable mode for SparkKubernetesOperator

Author: MaksYermak

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py

@@ -796,11 +796,13 @@ def _refresh_cached_properties(self):
         del self.pod_manager
 
     def execute_async(self, context: Context) -> None:
-        self.pod_request_obj = self.build_pod_request_obj(context)
-        self.pod = self.get_or_create_pod(  # must set `self.pod` for `on_kill`
-            pod_request_obj=self.pod_request_obj,
-            context=context,
-        )
+        if self.pod_request_obj is None:
+            self.pod_request_obj = self.build_pod_request_obj(context)
+        if self.pod is None:
+            self.pod = self.get_or_create_pod(  # must set `self.pod` for `on_kill`
+                pod_request_obj=self.pod_request_obj,
+                context=context,
+            )
         if self.callbacks:
             pod = self.find_pod(self.pod.metadata.namespace, context=context)
             for callback in self.callbacks:

providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/spark_kubernetes.py

@@ -268,8 +268,19 @@ def get_or_create_spark_crd(self, launcher: CustomObjectLauncher, context) -> k8
     def process_pod_deletion(self, pod, *, reraise=True):
         if pod is not None:
             if self.delete_on_termination:
-                self.log.info("Deleting spark job: %s", pod.metadata.name.replace("-driver", ""))
-                self.launcher.delete_spark_job(pod.metadata.name.replace("-driver", ""))
+                pod_name = pod.metadata.name.replace("-driver", "")
+                self.log.info("Deleting spark job: %s", pod_name)
+
+                # because of self.launcher is defined in execute method it does not exist for
+                # deferrable mode and for this reason launcher should be defined one more time
+                launcher = CustomObjectLauncher(
+                    name=pod_name,
+                    namespace=pod.metadata.namespace,
+                    kube_client=self.client,
+                    custom_obj_api=self.custom_obj_api,
+                    template_body=self.template_body,
+                )
+                launcher.delete_spark_job(pod_name)
             else:
                 self.log.info("skipping deleting spark job: %s", pod.metadata.name)