holmuk commented on code in PR #65058:
URL: https://github.com/apache/airflow/pull/65058#discussion_r3128486653
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/triggers/job.py:
##########
@@ -156,6 +191,206 @@ async def run(self) -> AsyncIterator[TriggerEvent]:
}
)
+ async def _wait_until_container_state_or_job_done(
+ self,
+ pod_name: str,
+ container_name: str,
+ wait_method: Any,
+ job_task: asyncio.Task[V1Job],
+ state_label: str,
+ ) -> WaitOutcome:
+ wait_task = asyncio.create_task(
+ wait_method(
+ name=pod_name,
+ namespace=self.pod_namespace,
+ container_name=container_name,
+ poll_interval=self.poll_interval,
+ )
+ )
+
+ try:
+ while not wait_task.done() and not job_task.done():
Review Comment:
Good catch, we may return `job_done` before `wait_task` gets CPU. However,
that is intentional behavior - once the `Job` is done, we stop strict container
waiting to avoid hangs and switch to bounded post-job best-effort XCom
extraction. The current pod is still included (`self.pod_names[pod_index:]`),
so it is not dropped from extraction attempts.
If you want, we can tighten semantics further by adding one immediate
non-blocking readiness probe before returning `job_done`.
--
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]
