Reconnect on read_pods when k8s auth fails

[kannon92]

closes: #32718

I have a patch from my client that they use to reconnect pods with KPO in the case of an expiring token.

Opening this up as is and happy to discuss better ideas!

Essentially we want to reconnect/reauthenticate on read_pods as we found that when the token expires our tasks start getting 401 errors.

[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 Contribution Guide (https://github.com/apache/airflow/blob/main/CONTRIBUTING.rst)
Here are some useful points:

• Pay attention to the quality of your code (ruff, mypy and type annotations). Our pre-commits 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.
  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

[kannon92]

@hussein-awala or @jedcunningham can I get CI approval to run the tests? I think I took some merges with the kubernetes cleanup and it caused some extra issues. I ran precommit and breeze and I think it should be good again.

[hussein-awala]

This change could fix your issue, but it will impact the performance of the KP operator and hook. I need to check how the Kubernetes python client handle the ExecCredentials and if there is another way or a a workaround to fix the issue without loading the client before each call. (for example through refresh_api_key_hook parameter).

And if not, maybe we can add a new parameter for client reloading and set it to False by default, and load the client only when it's True, something like:

- @cached_property
+ @property
def get_conn:
-  #prepare client
-  return client
+  if reload_client or not self.__cached_client:
+    self.__cached_client: = #prepare client
+  self.__cached_client:

[github-actions]

This pull request has been automatically marked as stale because it has not had recent activity. It will be closed in 5 days if no further activity occurs. Thank you for your contributions.