diff --git a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
index 90caa37428d6a..d093567bca1ee 100644
--- a/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
+++ b/airflow/providers/cncf/kubernetes/hooks/kubernetes.py
@@ -251,7 +251,6 @@ def create_custom_object(
     ):
         """
         Creates custom resource definition object in Kubernetes
-
         :param group: api group
         :param version: api version
         :param plural: api plural
@@ -260,26 +259,32 @@ def create_custom_object(
         """
         api = client.CustomObjectsApi(self.api_client)
         namespace = namespace or self._get_namespace() or self.DEFAULT_NAMESPACE
+
         if isinstance(body, str):
             body_dict = _load_body_to_dict(body)
         else:
             body_dict = body
-        try:
-            api.delete_namespaced_custom_object(
-                group=group,
-                version=version,
-                namespace=namespace,
-                plural=plural,
-                name=body_dict["metadata"]["name"],
-            )
-            self.log.warning("Deleted custom resource with the same name.")
-        except client.rest.ApiException:
-            self.log.info("Custom resource %s not found.", body_dict["metadata"]["name"])
+
+        # Attribute "name" is not mandatory if "generateName" is used instead
+        if "name" in body_dict["metadata"]:
+            try:
+                api.delete_namespaced_custom_object(
+                    group=group,
+                    version=version,
+                    namespace=namespace,
+                    plural=plural,
+                    name=body_dict["metadata"]["name"],
+                )
+
+                self.log.warning("Deleted SparkApplication with the same name")
+            except client.rest.ApiException:
+                self.log.info("SparkApplication %s not found", body_dict["metadata"]["name"])
 
         try:
             response = api.create_namespaced_custom_object(
                 group=group, version=version, namespace=namespace, plural=plural, body=body_dict
             )
+
             self.log.debug("Response: %s", response)
             return response
         except client.rest.ApiException as e:
diff --git a/tests/providers/cncf/kubernetes/operators/test_spark_kubernetes.py b/tests/providers/cncf/kubernetes/operators/test_spark_kubernetes.py
index d3e28b2539f5d..113562e21a4b9 100644
--- a/tests/providers/cncf/kubernetes/operators/test_spark_kubernetes.py
+++ b/tests/providers/cncf/kubernetes/operators/test_spark_kubernetes.py
@@ -1,4 +1,3 @@
-#
 # Licensed to the Apache Software Foundation (ASF) under one
 # or more contributor license agreements.  See the NOTICE file
 # distributed with this work for additional information
@@ -67,6 +66,49 @@
       - name: "test-volume"
         mountPath: "/tmp"
 """
+
+TEST_VALID_APPLICATION_YAML_USING_GENERATE_NAME = """
+apiVersion: "sparkoperator.k8s.io/v1beta2"
+kind: SparkApplication
+metadata:
+  generateName: spark-pi
+  namespace: default
+spec:
+  type: Scala
+  mode: cluster
+  image: "gcr.io/spark-operator/spark:v2.4.5"
+  imagePullPolicy: Always
+  mainClass: org.apache.spark.examples.SparkPi
+  mainApplicationFile: "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar"
+  sparkVersion: "2.4.5"
+  restartPolicy:
+    type: Never
+  volumes:
+    - name: "test-volume"
+      hostPath:
+        path: "/tmp"
+        type: Directory
+  driver:
+    cores: 1
+    coreLimit: "1200m"
+    memory: "512m"
+    labels:
+      version: 2.4.5
+    serviceAccount: spark
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+  executor:
+    cores: 1
+    instances: 1
+    memory: "512m"
+    labels:
+      version: 2.4.5
+    volumeMounts:
+      - name: "test-volume"
+        mountPath: "/tmp"
+"""
+
 TEST_VALID_APPLICATION_JSON = """
 {
    "apiVersion":"sparkoperator.k8s.io/v1beta2",
@@ -127,6 +169,7 @@
    }
 }
 """
+
 TEST_APPLICATION_DICT = {
     "apiVersion": "sparkoperator.k8s.io/v1beta2",
     "kind": "SparkApplication",
@@ -159,13 +202,53 @@
     },
 }
 
+TEST_APPLICATION_DICT_WITH_GENERATE_NAME = {
+    "apiVersion": "sparkoperator.k8s.io/v1beta2",
+    "kind": "SparkApplication",
+    "metadata": {"generateName": "spark-pi", "namespace": "default"},
+    "spec": {
+        "driver": {
+            "coreLimit": "1200m",
+            "cores": 1,
+            "labels": {"version": "2.4.5"},
+            "memory": "512m",
+            "serviceAccount": "spark",
+            "volumeMounts": [{"mountPath": "/tmp", "name": "test-volume"}],
+        },
+        "executor": {
+            "cores": 1,
+            "instances": 1,
+            "labels": {"version": "2.4.5"},
+            "memory": "512m",
+            "volumeMounts": [{"mountPath": "/tmp", "name": "test-volume"}],
+        },
+        "image": "gcr.io/spark-operator/spark:v2.4.5",
+        "imagePullPolicy": "Always",
+        "mainApplicationFile": "local:///opt/spark/examples/jars/spark-examples_2.11-2.4.5.jar",
+        "mainClass": "org.apache.spark.examples.SparkPi",
+        "mode": "cluster",
+        "restartPolicy": {"type": "Never"},
+        "sparkVersion": "2.4.5",
+        "type": "Scala",
+        "volumes": [{"hostPath": {"path": "/tmp", "type": "Directory"}, "name": "test-volume"}],
+    },
+}
+
 
 @patch("airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook.get_conn")
+@patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
+@patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
+@patch("airflow.utils.context.Context")
 class TestSparkKubernetesOperator(unittest.TestCase):
     def setUp(self):
         db.merge_conn(
-            Connection(conn_id="kubernetes_default_kube_config", conn_type="kubernetes", extra=json.dumps({}))
+            Connection(
+                conn_id="kubernetes_default_kube_config",
+                conn_type="kubernetes",
+                extra=json.dumps({}),
+            )
         )
+
         db.merge_conn(
             Connection(
                 conn_id="kubernetes_with_namespace",
@@ -173,13 +256,12 @@ def setUp(self):
                 extra=json.dumps({"namespace": "mock_namespace"}),
             )
         )
+
         args = {"owner": "airflow", "start_date": timezone.datetime(2020, 2, 1)}
         self.dag = DAG("test_dag_id", default_args=args)
 
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
     def test_create_application_from_yaml(
-        self, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
     ):
         op = SparkKubernetesOperator(
             application_file=TEST_VALID_APPLICATION_YAML,
@@ -187,8 +269,10 @@ def test_create_application_from_yaml(
             kubernetes_conn_id="kubernetes_default_kube_config",
             task_id="test_task_id",
         )
-        op.execute(None)
+
+        op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
+
         mock_delete_namespaced_crd.assert_called_once_with(
             group="sparkoperator.k8s.io",
             namespace="default",
@@ -196,6 +280,7 @@ def test_create_application_from_yaml(
             version="v1beta2",
             name=TEST_APPLICATION_DICT["metadata"]["name"],
         )
+
         mock_create_namespaced_crd.assert_called_with(
             body=TEST_APPLICATION_DICT,
             group="sparkoperator.k8s.io",
@@ -204,10 +289,30 @@ def test_create_application_from_yaml(
             version="v1beta2",
         )
 
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
+    def test_create_application_from_yaml_using_generate_name(
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+    ):
+        op = SparkKubernetesOperator(
+            application_file=TEST_VALID_APPLICATION_YAML_USING_GENERATE_NAME,
+            dag=self.dag,
+            kubernetes_conn_id="kubernetes_default_kube_config",
+            task_id="test_task_id",
+        )
+
+        op.execute(context)
+        mock_kubernetes_hook.assert_called_once_with()
+        mock_delete_namespaced_crd.assert_not_called()
+
+        mock_create_namespaced_crd.assert_called_with(
+            body=TEST_APPLICATION_DICT_WITH_GENERATE_NAME,
+            group="sparkoperator.k8s.io",
+            namespace="default",
+            plural="sparkapplications",
+            version="v1beta2",
+        )
+
     def test_create_application_from_json(
-        self, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
     ):
         op = SparkKubernetesOperator(
             application_file=TEST_VALID_APPLICATION_JSON,
@@ -215,8 +320,10 @@ def test_create_application_from_json(
             kubernetes_conn_id="kubernetes_default_kube_config",
             task_id="test_task_id",
         )
-        op.execute(None)
+
+        op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
+
         mock_delete_namespaced_crd.assert_called_once_with(
             group="sparkoperator.k8s.io",
             namespace="default",
@@ -224,6 +331,7 @@ def test_create_application_from_json(
             version="v1beta2",
             name=TEST_APPLICATION_DICT["metadata"]["name"],
         )
+
         mock_create_namespaced_crd.assert_called_with(
             body=TEST_APPLICATION_DICT,
             group="sparkoperator.k8s.io",
@@ -232,13 +340,12 @@ def test_create_application_from_json(
             version="v1beta2",
         )
 
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
     def test_create_application_from_json_with_api_group_and_version(
-        self, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
     ):
         api_group = "sparkoperator.example.com"
         api_version = "v1alpha1"
+
         op = SparkKubernetesOperator(
             application_file=TEST_VALID_APPLICATION_JSON,
             dag=self.dag,
@@ -247,8 +354,10 @@ def test_create_application_from_json_with_api_group_and_version(
             api_group=api_group,
             api_version=api_version,
         )
-        op.execute(None)
+
+        op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
+
         mock_delete_namespaced_crd.assert_called_once_with(
             group=api_group,
             namespace="default",
@@ -256,6 +365,7 @@ def test_create_application_from_json_with_api_group_and_version(
             version=api_version,
             name=TEST_APPLICATION_DICT["metadata"]["name"],
         )
+
         mock_create_namespaced_crd.assert_called_with(
             body=TEST_APPLICATION_DICT,
             group=api_group,
@@ -264,10 +374,8 @@ def test_create_application_from_json_with_api_group_and_version(
             version=api_version,
         )
 
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
     def test_namespace_from_operator(
-        self, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
     ):
         op = SparkKubernetesOperator(
             application_file=TEST_VALID_APPLICATION_JSON,
@@ -276,8 +384,10 @@ def test_namespace_from_operator(
             kubernetes_conn_id="kubernetes_with_namespace",
             task_id="test_task_id",
         )
-        op.execute(None)
+
+        op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
+
         mock_delete_namespaced_crd.assert_called_once_with(
             group="sparkoperator.k8s.io",
             namespace="operator_namespace",
@@ -285,6 +395,7 @@ def test_namespace_from_operator(
             version="v1beta2",
             name=TEST_APPLICATION_DICT["metadata"]["name"],
         )
+
         mock_create_namespaced_crd.assert_called_with(
             body=TEST_APPLICATION_DICT,
             group="sparkoperator.k8s.io",
@@ -293,10 +404,8 @@ def test_namespace_from_operator(
             version="v1beta2",
         )
 
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.delete_namespaced_custom_object")
-    @patch("kubernetes.client.api.custom_objects_api.CustomObjectsApi.create_namespaced_custom_object")
     def test_namespace_from_connection(
-        self, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
+        self, context, mock_create_namespaced_crd, mock_delete_namespaced_crd, mock_kubernetes_hook
     ):
         op = SparkKubernetesOperator(
             application_file=TEST_VALID_APPLICATION_JSON,
@@ -304,8 +413,10 @@ def test_namespace_from_connection(
             kubernetes_conn_id="kubernetes_with_namespace",
             task_id="test_task_id",
         )
-        op.execute(None)
+
+        op.execute(context)
         mock_kubernetes_hook.assert_called_once_with()
+
         mock_delete_namespaced_crd.assert_called_once_with(
             group="sparkoperator.k8s.io",
             namespace="mock_namespace",
@@ -313,6 +424,7 @@ def test_namespace_from_connection(
             version="v1beta2",
             name=TEST_APPLICATION_DICT["metadata"]["name"],
         )
+
         mock_create_namespaced_crd.assert_called_with(
             body=TEST_APPLICATION_DICT,
             group="sparkoperator.k8s.io",
