K8S Spark Operator doesn't delete an older application with the same name anymore

[assaf-xm]

Apache Airflow version

2.6.2

What happened

Changes to the SparkKubernetesOperator which introduced in #21092 were reverted by the changes added in #29977

Until 2.6.2 when a spark application was created with the same name of a previous application, the old application (pods etc') was automatically deleted and the new application started (with the warning of 'Deleted SparkApplication with the same name' in the logs).
After 2.6.2 this scenario is causing an error starting the new application (409 - Already exists) and the previous application needs to be manually deleted (there is no easy way to delete it programmatically from airflow dags).

Was this behavior change done on purpose?
I can suggest to add a parameter to control the desired behavior.

What you think should happen instead

Changes to the SparkKubernetesOperator which introduced in #21092 were reverted by the changes added in #29977

How to reproduce

Start a new spark application with the same name of the previous one, compare 2.6.2 vs 2.5.3

Operating System

EKS

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

• Yes I am willing to submit a PR!

Code of Conduct

• I agree to follow this project's Code of Conduct

[boring-cyborg]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[nathadfield]

@assaf-xm It seems like this ability was deemed out of scope for Airflow as per the PR comments. As the commiiter, perhaps @blcksrx could give some more context?

[blcksrx]

Well if you look closely to the older PR you would find the delete operation defined in the KubernetesHook and also it produces logs about SparkApplication beside of this fact that creating_namespaced_object is not only about creating the SparkApplications, it gives away the single responsibility principal of the method createing_namespaced_object.
In addition, Since it deletes that namespaced_object, it rewrites the object history in k8s and it can not be understand why that object failed for the first time!

 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"])

It's better to use simple macros on the object name such as name-{{ ds }} and that would solved the problem

[assaf-xm]

@blcksrx , note that the removal of this code breaks applications that use a constant names, so having the old behavior as an option will be very useful for some of the users.
Working with name-{{ ds }} could be a solution, however I find it less attractive since it can 'pollute' with a huge number of old spark applications and also finding the last spark application that ran is harder this way.

[blcksrx]

@assaf-xm from the compatibility, yes you are absolutely right, on the other hand, developer should do the cleaning by it self

[assaf-xm]

@blcksrx , is there any easy way to clean spark applications from airflow dags?
In any case, please consider adding this as an optional parameter to SparkKubernetesOperator

[nathadfield]

@assaf-xm You are welcome to submit a PR for this and reference the conversation above at which point it will be reviewed and decided upon. Otherwise this will remain as an open "issue" unless someone else picks it up.