Make Kubernetes Executor & Scheduler Resilient to error during PMH execution

[XD-DENG]

Currently there is no try-catch logic when pod_mutation_hook is called in the Kubernetes Executor. So if there is any error/exception during the execution of the pod_mutation_hook user created, the whole executor/scheduler will crash.

Such error may happen:

• user who authored the pod_mutation_hook made an error inside.
• if there is any code inside the pod_mutation_hook encounters transient error (e.g. it queries DB and the DB gives a timeout error).
• ... you name it. It can simply go wrong.

This PR aims to make Kubernetes Executor & Scheduler more resilient to such error during PMH execution (current behavior, directly crashing, is a bit scary)

---

^ Add meaningful description above

Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[hterik]

I believe this should be raise PodMutationHookException() from e ? Otherwise it will be logged as an exception happening while handling another exception.

[XD-DENG]

Hi @hterik , thanks for the feedback!

I tested the two different ways below. I find the difference is minor, and I personally prefer the current way where the detailed exception reason appears in the same line as the PodMutationHookException, so it's clearer to the log reader (this is also the original reason for what I wrote it this way). WDYT?

Case -1

from airflow.exceptions import PodMutationHookException

try:
    int("a")
except Exception as e:
    raise PodMutationHookException(e)

We get

Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException: invalid literal for int() with base 10: 'a'

Case -2

If we use the way you suggested

from airflow.exceptions import PodMutationHookException

try:
    int("a")
except Exception as e:
    raise PodMutationHookException() from e

We get

Traceback (most recent call last):
  File "<stdin>", line 2, in <module>
ValueError: invalid literal for int() with base 10: 'a'

The above exception was the direct cause of the following exception:

Traceback (most recent call last):
  File "<stdin>", line 4, in <module>
airflow.exceptions.PodMutationHookException

[potiuk]

Better indeed.

[hterik]

Oh sorry, maybe raise PodMutationHookException(e) from e in that case ;) That will have best of both worlds. But yeah, i think the difference is minor and i don't think it has any logical difference other than the log message in between the two tracebacks.