arkadiuszbach commented on code in PR #69058:
URL: https://github.com/apache/airflow/pull/69058#discussion_r3493390194
##########
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:
Kubernetes may generate multiple events with `pod.status.phase=Failed` in a
short period of time for a single pod. I was checking recently what events it
would generate due to SIGTERM and noticed two `MODIFIED` followed by `DELETED`,
all `pod.status.phase=Failed`, see below:
```
2026-06-29T15:26:17.051533Z [debug ] Event: test-kubernetes-0-l32duhyu
had an event of type MODIFIED
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:159
2026-06-29T15:26:17.051986Z [debug ] Creating task key for annotations
{'dag_id': 'test', 'task_id': 'kubernetes_0', 'logical_date': None, 'run_id':
'manual__2026-06-25T14:26:59.504591+00:00', 'try_number': '3'}
[airflow.providers.cncf.kubernetes.kubernetes_helper_functions]
loc=kubernetes_helper_functions.py:159
2026-06-29T15:26:17.052418Z [warning ] Event: test-kubernetes-0-l32duhyu
Failed, task: test.kubernetes_0.3, annotations: <omitted>
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:308
2026-06-29T15:26:17.340492Z [debug ] Event: test-kubernetes-0-l32duhyu
had an event of type MODIFIED
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:159
2026-06-29T15:26:17.340786Z [debug ] Creating task key for annotations
{'dag_id': 'test', 'task_id': 'kubernetes_0', 'logical_date': None, 'run_id':
'manual__2026-06-25T14:26:59.504591+00:00', 'try_number': '3'}
[airflow.providers.cncf.kubernetes.kubernetes_helper_functions]
loc=kubernetes_helper_functions.py:159
2026-06-29T15:26:17.340985Z [warning ] Event: test-kubernetes-0-l32duhyu
Failed, task: test.kubernetes_0.3, annotations: <omitted>
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:308
2026-06-29T15:26:17.372148Z [debug ] Running
SchedulerJobRunner._create_dagruns_for_dags with retries. Try 1 of 50
[airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=retries.py:100
2026-06-29T15:26:17.381234Z [debug ] Event: test-kubernetes-0-l32duhyu
had an event of type DELETED
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:159
2026-06-29T15:26:17.381554Z [debug ] Creating task key for annotations
{'dag_id': 'test', 'task_id': 'kubernetes_0', 'logical_date': None, 'run_id':
'manual__2026-06-25T14:26:59.504591+00:00', 'try_number': '3'}
[airflow.providers.cncf.kubernetes.kubernetes_helper_functions]
loc=kubernetes_helper_functions.py:159
2026-06-29T15:26:17.381744Z [warning ] Event: test-kubernetes-0-l32duhyu
Failed, task: test.kubernetes_0.3, annotations: <omitted>
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher]
loc=kubernetes_executor_utils.py:308
```
so maybe `self.last_known_jobs.pop(key, None)` should be placed right after
`self.task_queue.put(job)`?
##########
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:
Seems like this is currently a dead code, TaskInstanceState.RUNNING is never
set in
[process_status](https://github.com/apache/airflow/blob/e246e8fb37740a6a988e41ed79655a05c518d075/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor_utils.py#L326)
From what i can understand it will indicate POD running status, but POD may
get Failed after being RUNNING and the task still being in QUEUED state, for
example due to SIGTERM
--
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]
