seanmuth commented on code in PR #69058:
URL: https://github.com/apache/airflow/pull/69058#discussion_r3500326897
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -455,13 +471,17 @@ def _change_state(
if state == ADOPTED:
# When the task pod is adopted by another executor,
# then remove the task from the current executor running queue.
+ self.last_known_jobs.pop(key, None)
try:
self.running.remove(key)
except KeyError:
self.log.debug("TI key not in running: %s", key)
return
if state == TaskInstanceState.RUNNING:
Review Comment:
Right — removed the cleanup I'd added in the `RUNNING` branch, since the
watcher never emits `RUNNING`. On the SIGTERM case you raised: a pod that
reaches Running then Fails before the runner writes `running` stays `QUEUED`,
so the pre-execution path requeues it — which is the intended outcome (the task
never committed to running), and complements #69034.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @seanmuth before posting
##########
providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py:
##########
@@ -478,6 +498,37 @@ def _change_state(
self.kube_scheduler.patch_pod_executor_done(pod_name=pod_name,
namespace=namespace)
self.log.info("Patched pod %s in namespace %s to mark it as done",
key, namespace)
+ if state == TaskInstanceState.FAILED and
self._is_pre_execution_failure(
+ state,
+ self._get_task_instance_state(key, session=session),
+ failure_details,
+ self.pod_launch_failure_excluded_container_reasons,
+ ):
+ attempts = self.pod_launch_failure_attempts[key]
+ job = self.last_known_jobs.get(key)
+ can_requeue = (
+ self.pod_launch_failure_max_retries == -1 or attempts <
self.pod_launch_failure_max_retries
+ )
+ if can_requeue and job is not None:
+ self.pod_launch_failure_attempts[key] = attempts + 1
+ self.log.warning(
+ "[Try %s of %s] Pod %s/%s for task %s failed before the
task process started "
+ "(container_reason: %s). Requeuing without consuming a
task retry.",
+ attempts + 1,
+ self.pod_launch_failure_max_retries,
+ namespace,
+ pod_name,
+ key,
+ failure_details.get("container_reason") if failure_details
else None,
+ )
+ # Leave the key in self.running and do not write to
event_buffer: the scheduler
+ # never observes this failure, so no task-level retry is
consumed.
+ self.task_queue.put(job)
Review Comment:
Good catch — confirmed (MODIFIED×2 + DELETED on SIGTERM). I went a slightly
different way than popping the job right after `put`: that would cap requeues
at 1, because the requeue path re-queues the stored job directly rather than
going back through `execute_async` to re-stash it. Instead the executor records
the pod each requeue was issued for (`_PodLaunchAttempt.requeued_for_pod`) and
ignores duplicate `Failed` events for that same pod, while a distinct
(requeued) pod still requeues. Covered by
`test_change_state_pre_execution_failure_dedupes_repeated_events`. Fixed in
`2352e1f453`.
---
Drafted-by: Claude Code (Opus 4.8); reviewed by @seanmuth before posting
--
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]
