shahar1 commented on code in PR #63915:
URL: https://github.com/apache/airflow/pull/63915#discussion_r3115406005
##########
providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py:
##########
@@ -333,9 +333,24 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
f"package
{kubernetes_provider_name}=={kubernetes_provider_version} which doesn't "
f"support this feature. Please upgrade it to version
higher than or equal to {min_version}."
)
+ # Re-discover pods for this job to handle retry scenarios
+ # where the original pod failed and a new pod was created
+ label_selector = f"job-name={self.job_name}"
+ current_pods = await self.hook.list_pods(
+ namespace=self.pod_namespace,
+ label_selector=label_selector,
+ )
+ succeeded_pods = [p for p in current_pods if p.status and
p.status.phase == "Succeeded"]
+ if not succeeded_pods:
Review Comment:
Maybe we should wrap it in an `while True` loop (+timeout) rather an if
statment?
Otherwise, there are scenarios where this logic can still fail, for example
if we query too early and no pod is Succeeded yet, we fall back to the original
pod list and can miss the replacement retry pod that later succeeds and
contains the real XCom.
##########
providers/google/src/airflow/providers/google/cloud/triggers/kubernetes_engine.py:
##########
@@ -333,9 +333,24 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
f"package
{kubernetes_provider_name}=={kubernetes_provider_version} which doesn't "
f"support this feature. Please upgrade it to version
higher than or equal to {min_version}."
)
+ # Re-discover pods for this job to handle retry scenarios
+ # where the original pod failed and a new pod was created
+ label_selector = f"job-name={self.job_name}"
+ current_pods = await self.hook.list_pods(
+ namespace=self.pod_namespace,
+ label_selector=label_selector,
+ )
+ succeeded_pods = [p for p in current_pods if p.status and
p.status.phase == "Succeeded"]
+ if not succeeded_pods:
Review Comment:
Maybe we should wrap it in an `while True` loop (+timeout) rather an if
statment?
Otherwise, there are scenarios where this logic can still fail, for example
- if we query too early and no pod is Succeeded yet, we fall back to the
original pod list and can miss the replacement retry pod that later succeeds
and contains the real XCom.
--
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]
