This is an automated email from the ASF dual-hosted git repository.
kaxil pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 8e14de67e3f Fix `KubernetesExecutor` leaving invalid-config tasks
stuck in queued (#74512)
8e14de67e3f is described below
commit 8e14de67e3f18c9f97d73e11fb2636f267b764e2
Author: Kevin Yang <[email protected]>
AuthorDate: Fri Oct 9 17:11:57 2026 -0400
Fix `KubernetesExecutor` leaving invalid-config tasks stuck in queued
(#74512)
When execute_async rejected a task's executor_config, it failed the task
before the key was recorded as running, and the key was then added to
running anyway. The scheduler saw the executor still owning the task and
ignored the FAILED event, so the task instance stayed queued until
task_queued_timeout and its parallelism slot was never released.
---
.../kubernetes/executors/kubernetes_executor.py | 3 +-
.../executors/test_kubernetes_executor.py | 40 ++++++++++++++++++++++
2 files changed, 42 insertions(+), 1 deletion(-)
diff --git
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
index 8c2477bfd57..0534f10799a 100644
---
a/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
+++
b/providers/cncf/kubernetes/src/airflow/providers/cncf/kubernetes/executors/kubernetes_executor.py
@@ -434,8 +434,9 @@ class KubernetesExecutor(BaseExecutor):
del self.executor_queues[WorkloadType.EXECUTE_TASK][key]
else:
del self.queued_tasks[key]
- self.execute_async(key=key, command=command, queue=queue,
executor_config=executor_config)
+ # Add to running before execute_async: it may fail() the task,
which must remove the key again.
self.running.add(key)
+ self.execute_async(key=key, command=command, queue=queue,
executor_config=executor_config)
def _should_create_pod_for_job(self, task: KubernetesJob) -> bool:
"""
diff --git
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
index fc08f0338c7..26d0aa254d6 100644
---
a/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
+++
b/providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/test_kubernetes_executor.py
@@ -1772,6 +1772,46 @@ class TestKubernetesExecutor:
assert key not in executor.running
assert executor.get_event_buffer() == {key:
(TaskInstanceState.SUCCESS, None)}
+ @pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test requires Airflow
3+")
+ def
test_process_workloads_invalid_executor_config_is_not_tracked_as_running(self):
+ """A task failed by execute_async must not stay in ``running``.
+
+ Otherwise ``has_task`` keeps returning True and the scheduler ignores
the FAILED event,
+ leaving the task instance stuck in queued until
``task_queued_timeout``.
+ """
+ executor = self.kubernetes_executor
+ workload = ExecuteTask(
+ ti=WorkloadTaskInstance(
+ id=uuid4(),
+ dag_version_id=uuid4(),
+ dag_id="dag",
+ task_id="task",
+ run_id="run_id",
+ try_number=1,
+ map_index=-1,
+ pool_slots=1,
+ priority_weight=1,
+ queue="default",
+ executor_config={"KubernetesExecutor": {"config_file":
"/some/path/kubeconfig.yaml"}},
+ ),
+ dag_rel_path="dag.py",
+ bundle_info=BundleInfo(name="bundle"),
+ token="",
+ log_path=None,
+ )
+ key = executor.get_task_key(workload.ti) if
executor.supports_task_instance_uuid else workload.ti.key
+ if AIRFLOW_V_3_1_PLUS:
+ executor.queue_workload(workload, session=None)
+ else:
+ executor.queue_command(workload.ti, [workload],
workload.ti.priority_weight, workload.ti.queue)
+
+ executor._process_workloads([workload])
+
+ assert key not in executor.running
+ assert executor.get_event_buffer() == {
+ key: (TaskInstanceState.FAILED, "Invalid executor_config passed")
+ }
+
@pytest.mark.skipif(not AIRFLOW_V_3_0_PLUS, reason="Test requires Airflow
3+")
def test_queue_workload_queues_execute_task(self):
"""queue_workload must queue an ExecuteTask on every supported Airflow
3 version.
