Add KubernetesPodOperatorAsync

[phanikumv]

This PR donates the KubernetesPodOperatorAsync from astronomer-providers repo to Airflow.

Astronomer and its customers have been running this in production since last 6 months

---

^ 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.

[dimberman]

Overall this looks good to me. My only major question is whether we need to asynchronously load the configuration. otherwise great job!

[dimberman]

Is this function really one that requires async? I imagine loading a kube config would be pretty quick?

[dstandish]

you're referring to async def _load_config(self) right?

well it does do file operations which are technically blocking (even though they might be fast)

so you might as well handle them asyncly ... just cus... why not?

but now that you mention it one thing that is a bit curious is ... how are we able to await await config.load_kube_config becaus it look like that is not an async method... i must be missing something

[kaxil]

yup, @phanikumv this config should come from from kubernetes_asyncio import config instead of from kubernetes import config like it is doing right now.

Please check https://github.com/astronomer/astronomer-providers/blob/main/astronomer/providers/cncf/kubernetes/hooks/kubernetes.py

[VladaZakharova]

May be place more descriptive comment here to describe what does actually in_cluster parameter mean?

[dstandish]

you're referring to async def _load_config(self) right?

well it does do file operations which are technically blocking (even though they might be fast)

so you might as well handle them asyncly ... just cus... why not?

but now that you mention it one thing that is a bit curious is ... how are we able to await await config.load_kube_config becaus it look like that is not an async method... i must be missing something

[kaxil]

#28230 was created first, lets get that one merged and than rebase this on top of it to add more features

[phanikumv]

#28230 was created first, lets get that one merged and than rebase this on top of it to add more features

I am working to rebase it and add more features

[phanikumv]

Closing as the additional feature on periodic fetching of logs has been implemented in #27758