amoghrajesh commented on code in PR #69914:
URL: https://github.com/apache/airflow/pull/69914#discussion_r3612720916
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -610,9 +640,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:
Handled in 8335567339
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/operators/pod.py:
##########
@@ -377,6 +391,18 @@ def __init__(
log_formatter: Callable[[str, str], str] | None = None,
**kwargs,
) -> None:
+ if "reattach_on_restart" in kwargs:
Review Comment:
Handled in 8335567339
--
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]
