Airflow task takes success status if driver was deleted

[andallo]

Apache Airflow Provider(s)

cncf-kubernetes

Versions of Apache Airflow Providers

apache-airflow-providers-cncf-kubernetes 8.3.1

Apache Airflow version

2.9.2

Operating System

Debian GNU/Linux 12 (bookworm)

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

What happened

If driver pod was deleted (manually), SparkKubernetesOperator task take 'success' status, despite driver wasn't complete all the jobs of SparkApplication.

What you think should happen instead

If driver pod was deleted, but wasn't complete his work, SparkKubernetesOperator task should take 'failed' status.

How to reproduce

Start SparkApplication using SparkKubernetesOperator task. Wait until driver pod will be created. Delete driver pod during it does some work. SparkApplication completes immediately (with status=COMPLETED) and task will become success, despite driver wasn't complete the work.

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

[bugraoz93]

I have started working on a fix. I will publish the PR soon

[bugraoz93]

Have you experienced any similar issues with your KubernetesPodOpertors or any operator from the provider?

[andallo]

I haven't been deal with other operators, only with SparkKubernetesOperator.

Let me give some notes about PR. I tested changes of custom_object_launcher.py and result isn't what I expect. First launch of the application failed with error:

[2024-08-05, 06:33:26 UTC] {taskinstance.py:2905} ERROR - Task failed with exception
Traceback (most recent call last):
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 465, in _execute_task
    result = _execute_callable(context=context, **execute_callable_kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 432, in _execute_callable
    return execute_callable(context=context, **execute_callable_kwargs)
  File "/home/airflow/.local/lib/python3.8/site-packages/airflow/models/baseoperator.py", line 401, in wrapper
    return func(self, *args, **kwargs)
  File "/opt/airflow/dags/repo/custom_operators/spark_kubernetes_operator_8_3_1.py", line 314, in execute
    self.pod = self.get_or_create_spark_crd(self.launcher, context)
  File "/opt/airflow/dags/repo/custom_operators/spark_kubernetes_operator_8_3_1.py", line 250, in get_or_create_spark_crd
    driver_pod, spark_obj_spec = launcher.start_spark_job(
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 330, in wrapped_f
    return self(f, *args, **kw)
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 467, in __call__
    do = self.iter(retry_state=retry_state)
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 368, in iter
    result = action(retry_state)
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 390, in <lambda>
    self._add_action_func(lambda rs: rs.outcome.result())
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 437, in result
    return self.__get_result()
  File "/usr/local/lib/python3.8/concurrent/futures/_base.py", line 389, in __get_result
    raise self._exception
  File "/home/airflow/.local/lib/python3.8/site-packages/tenacity/__init__.py", line 470, in __call__
    result = fn(*args, **kwargs)
  File "/opt/airflow/dags/repo/custom_operators/custom_object_launcher.py", line 313, in start_spark_job
    raise e
  File "/opt/airflow/dags/repo/custom_operators/custom_object_launcher.py", line 300, in start_spark_job
    while self.spark_job_not_running(self.spark_obj_spec):
  File "/opt/airflow/dags/repo/custom_operators/custom_object_launcher.py", line 369, in spark_job_not_running
    raise AirflowException(
airflow.exceptions.AirflowException: Spark Job Status/Driver State not found. Please check if the Job/Pod is running.

But this error failed only Airflow task (pod). Driver and spark application were continue to work. Second try started the task, because reattach_on_restart logic connected new task to the driver and CustomLauncher wasn't used. Then I tried to remove Driver pod manually and I was expected the task became failed. But again task took success status.

I think you try to solve the case when Driver are deleted during the launching of application. But I described different case in this issue. I delete Driver after application is launched, when driver is running and application already returned from start_spark_job function of CustomLauncher.

[bugraoz93]

Thanks for sharing more information!
Yes, this is the before part indeed. I just saw a case for the deleted pod in the Spark job submit phase and tried to include it in the solution.
From here this operator is using KubernetesPodOperator flavours. After it launches the driver pod, the remaining part of the operator is managed by KubernetesPodOperator. That's why I asked if you are using it and experiencing similar results where you are killing the pod but the Airflow is showing your task as a success.
I will check this more during the weekend and hope I can get back to you with a solution

[bugraoz93]

I haven't had the chance to check this yet, but I plan to do so in the next few days.

[bugraoz93]

Okay, I checked the same case in the main branch. I couldn't reproduced the same result :( When the task submitted and in running state, I have deleted the driver pod manually. My task went into up_for_retry. I have got a proper error message which is stating pod not found raise ApiException(http_resp=r) kubernetes.client.exceptions.ApiException: (404) Reason: Not Found
Could you please use later versions? Maybe 2.10.0, it's recently released and should have updated code.