cncf-kubernetes: fix potential race condition in trigger_reentry flow caused by is_istio_enabled

[noah-gil]

Problem

When using the deferred KubernetesPodOperator, there is a potential race condition in the trigger_reentry success flow. After a pod has completed and the worker resumes the workload from the triggerer, it reads the pod from the Kubernetes API, checking that it still exists:

try:
    self.pod = self.hook.get_pod(pod_name, pod_namespace)
except ApiException as e:
    if e.status != 404:
        raise

It then continues down the happy path until it reaches this part in _clean:

if self.pod is None:
    return

istio_enabled = self.is_istio_enabled(self.pod)

Then it is_istio_enabled, it attempts a second pod read from the Kubernetes API:

remote_pod = self.pod_manager.read_pod(pod)

At this point, it is possible that the pod has now been deleted by some external factor during the time between the first and second pod read. Should this occur, the task fails due to a raised ApiException error. Even though the pod succeeded, it is marked as error due to this failed API read, exactly what the first read tried to check for.

Solution

I chose to mimic the existing case of returning False when there is no pod provided in the function's argument. It narrowly catches the 404 and returns false, while still re-raising an exception for any other API error. I am open for discussion on other approaches for this fix. As this is quite a small patch, do let me know if I should create any tests to test for this exact scenario.

Thank you for your time!

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)

---

[boring-cyborg]

Congratulations on your first Pull Request and welcome to the Apache Airflow community! If you have any issues or are unsure about any anything please check our Contributors' Guide
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our prek-hooks will help you with that.
• In case of a new feature add useful documentation (in docstrings or in docs/ directory). Adding a new operator? Check this short guide Consider adding an example Dag that shows how users should use it.
• Consider using Breeze environment for testing locally, it's a heavy docker but it ships with a working Airflow and a lot of integrations.
• Be patient and persistent. It might take some time to get a review or get the final approval from Committers.
• Please follow ASF Code of Conduct for all communication including (but not limited to) comments on Pull Requests, Mailing list and Slack.
• Be sure to read the Airflow Coding style.
• Always keep your Pull Requests rebased, otherwise your build might fail due to changes not related to your commits.
  Apache Airflow is a community-driven project and together we are making it better 🚀.
  In case of doubts contact the developers at:
  Mailing List: dev@airflow.apache.org
  Slack: https://s.apache.org/airflow-slack

[Miretpl]

When using the deferred KubernetesPodOperator, there is a potential race condition in the trigger_reentry success flow.

Not only here, as far as I checked, but in general yes, that is a really nice catch

[Miretpl]

I will wait a day or two to merge if anyone also wants to provide feedback on it.