amoghrajesh commented on code in PR #69914:
URL: https://github.com/apache/airflow/pull/69914#discussion_r3622289843
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -614,9 +643,53 @@ def log_matching_pod(self, pod: k8s.V1Pod, context:
Context) -> None:
self.log.info("`try_number` of task_instance: %s",
context["ti"].try_number)
self.log.info("`try_number` of pod: %s",
pod.metadata.labels["try_number"])
+ def _get_pod_from_task_state_store(self, context: Context) -> k8s.V1Pod |
None:
+ task_state_store = context.get("task_state_store")
+ if task_state_store is None:
+ return None
+ stored = task_state_store.get(POD_IDENTIFIER_STATE_KEY)
+ if not isinstance(stored, dict):
+ return None
+ name, namespace = stored.get("name"), stored.get("namespace")
+ if not isinstance(name, str) or not isinstance(namespace, str):
+ self.log.warning(
+ "Pod identity stored in task state store is malformed (%s),
falling back to label search.",
+ stored,
+ )
+ return None
+ try:
+ pod = self.hook.get_pod(name, namespace)
Review Comment:
Fixed: `_get_pod_from_task_state_store` now checks the fetched pods
already_checked label and if a prior attempt already marked it, it's treated as
not reattachable and falls through to fresh pod creation instead of handing
back a dead pod from a prior attempt.
Added a test covering this too. (FAILED + already_checked=True stored pod,
on_finish_action="keep_pod") -- verified it fails without the fix and passes
with it. Fixed in 0218939c74
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
