sjyangkevin opened a new pull request, #74512:
URL: https://github.com/apache/airflow/pull/74512
### Why
closes: #67643
When a task's `executor_config` can't be converted to a pod (for example the
legacy `{"KubernetesExecutor": {...}}` dict), `KubernetesExecutor` reports the
task as failed, but the scheduler ignores that event. The task instance stays
`queued` until `[scheduler] task_queued_timeout` cleans it up, and one executor
parallelism slot leaks for the life of the scheduler process. With this change
the task is marked failed right away.
### Root cause
`running` means "this executor owns the task and hasn't reported its final
state yet". Every exit path removes the key from `running` at the same time it
writes the final state to `event_buffer` (`fail()` / `success()` /
`change_state()` / `_change_state()`). `_process_workloads()` broke that rule
by recording ownership after dispatch:
```python
self.execute_async(...) # invalid config -> self.fail(key): removes
nothing, buffers FAILED
self.running.add(key) # re-adds a key that has already been reported as
failed
```
When the scheduler processes the FAILED event, `executor.has_task(ti)`
returns `True` because the key is in `running`. The scheduler then treats the
event as "task requeued" (`scheduler_job_runner.py`, `ti_requeued`) and skips
failure handling. Nothing ever removes the key, because no pod was created, so
there's no watcher event.
### Fix
Add the key to `running` before calling `execute_async()`. A `fail()` inside
`execute_async()` then removes it through the existing `change_state()` path.
Pod-creation failures in `sync()` already work this way: they call `fail()`
while the key is in
`running`.
- The success path is unchanged. `_process_workloads()` runs synchronously
in the scheduler thread, and after it returns `running` holds the same keys as
before. The existing `test_process_workloads` passes unchanged.
- Any current or future `fail(); return` inside `execute_async()` is covered
without further changes.
- Trade-off: if `execute_async()` raised an unexpected exception (not a
handled `fail()`), the key would now be in `running` while the exception
propagates. That exception already aborts the scheduler loop, and executor
state is rebuilt through adoption on restart, so this doesn't make things worse.
### Tests
- New regression test
`test_process_workloads_invalid_executor_config_is_not_tracked_as_running`: it
runs an
`ExecuteTask` with the legacy config through `_process_workloads()`, using
the real
`execute_async()`. It asserts that the FAILED event is buffered and that
the key is not
in `running`. It fails on `main` and passes with the fix. The existing
`test_invalid_executor_config` calls `execute_async()` directly, which is
why it never
caught this.
Results:
- `providers/cncf/kubernetes/tests/unit/cncf/kubernetes/executors/`: 266
passed, 14 skipped
- `prek run --stage pre-commit --files <changed files>`: all hooks passed
- `breeze k8s tests -- -k test_kubernetes_executor` on kind
(KubernetesExecutor, fixed
image): 9 passed
### Reproduction on kind (KubernetesExecutor)
Dag from the issue:
```python
with DAG("test_k8s_executor_config_legacy", schedule=None, catchup=False):
PythonOperator(
task_id="check_for_failure",
python_callable=_noop,
executor_config={"KubernetesExecutor": {"config_file":
"/some/path/kubeconfig.yaml"}},
)
```
**Before (main):** the task is reported failed but stays queued, and the key
leaks in `running`:
```
04:03:15.497 [error] Invalid executor_config for 01a11ed4-...
Executor_config: {'KubernetesExecutor': {...}}
04:03:15.498 [info ] Received executor event with state failed for task
instance 01a11ed4-...
04:03:15.513 [info ] TaskInstance Finished: ... state=queued,
executor_state=failed
04:03:39.646 [debug] self.running:
{TaskInstanceUuid(id=UUID('01a11ed4-...'))}
```
No worker pod was created, there was no "state mismatch" handling, and the
task was
still `queued` more than 10 minutes later.
**After (this PR):** the scheduler reconciles the event and fails the task
immediately:
```
04:33:15.129 [error] Invalid executor_config for 01a11eef-...
Executor_config: {'KubernetesExecutor': {...}}
04:33:15.189 [info ] Received executor event with state failed for task
instance 01a11eef-...
04:33:15.199 [error] Executor KubernetesExecutor(parallelism=32) reported
that the task instance ... finished with state failed, but the task instance's
state attribute is queued. ...
04:33:15.203 [info ] Marking task as FAILED.
dag_id=test_k8s_executor_config_legacy, task_id=check_for_failure, ...
04:33:15.232 [info ] Marking run <DagRun test_k8s_executor_config_legacy
...> failed
```
The task UUID is no longer in `self.running`.
<details>
<summary>Commands used</summary>
```bash
breeze k8s deploy-cluster
breeze k8s deploy-airflow --executor KubernetesExecutor
export
KUBECONFIG=$PWD/.build/.k8s-clusters/airflow-python-3.12-v1.31.12/.kubeconfig
afs() { kubectl -n airflow exec deploy/airflow-scheduler -c scheduler -- \
env AIRFLOW__LOGGING__LOGGING_LEVEL=ERROR "$@"; }
DP=$(kubectl -n airflow get pod -l component=dag-processor -o name | head -1)
kubectl -n airflow cp test_k8s_executor_config_legacy.py \
${DP#pod/}:/opt/airflow/dags/test_k8s_executor_config_legacy.py -c
dag-processor
afs airflow dags unpause test_k8s_executor_config_legacy
afs airflow dags trigger test_k8s_executor_config_legacy -r repro_after_fix
afs airflow tasks states-for-dag-run test_k8s_executor_config_legacy
repro_after_fix -o json
kubectl -n airflow logs deploy/airflow-scheduler -c scheduler --since=5m \
| grep -E "Invalid executor_config|Received executor event|reported that
the task instance|Marking task as FAILED|self.running"
breeze k8s tests -- -k test_kubernetes_executor
```
</details>
---
##### Was generative AI tooling used to co-author this PR?
- [X] Yes (please specify the tool below)
Generated-by: Claude Code (Opus 5.5) following [the
guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#gen-ai-assisted-contributions)
--
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]
