67643 fix queued task issue in k8s operator

[kunaljubce]

• closes: KubernetesExecutor: Task stuck in queued state when using deprecated executor_config dict format #67643

So when I look into the scheduler logs after the task gets hung up in "queued state", I see the below:

2026-06-01T04:32:46.933008Z [error    ] Invalid executor_config for TaskInstanceKey(dag_id='test_k8s_executor_config_legacy', task_id='check_for_failure', run_id='manual__2026-06-01T04:32:45.257995+00:00',
try_number=1, map_index=-1). Executor_config: {'pod_override': 'something something', 'KubernetesExecutor': {'config_file': '/some/path/kubeconfig.yaml'}}
[airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:221
2026-06-01T04:32:46.933787Z [info     ] Received executor event with state failed for task instance TaskInstanceKey(dag_id='test_k8s_executor_config_legacy', task_id='check_for_failure',
run_id='manual__2026-06-01T04:32:45.257995+00:00', try_number=1, map_index=-1) [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1251
2026-06-01T04:32:46.944336Z [info     ] TaskInstance Finished: dag_id=test_k8s_executor_config_legacy, task_id=check_for_failure, run_id=manual__2026-06-01T04:32:45.257995+00:00, map_index=-1,
ti_id=019e8174-a2da-79ca-a4e8-bb0db76adec3, run_start_date=None, run_end_date=None, run_duration=None, state=queued, executor=KubernetesExecutor(parallelism=32), executor_state=failed, try_number=1, max_tries=0,
pool=default_pool, queue=default, priority_weight=1, operator=PythonOperator, queued_dttm=2026-06-01 04:32:46.925651+00:00, scheduled_dttm=2026-06-01 04:32:46.914614+00:00,queued_by_job_id=3, pid=None
[airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1340

However the task keeps being run within the scheduler as evident from the below logs:

2026-06-01T04:40:56.256451Z [debug    ] number of scheduleable tasks for <DagRun test_k8s_executor_config_legacy @ 2026-06-01 04:32:44+00:00: manual__2026-06-01T04:32:45.257995+00:00, state:running, queued_at:
2026-06-01 04:32:45.264255+00:00. run_type: manual>: 0 task(s) [airflow.models.dagrun.DagRun] loc=dagrun.py:1304
2026-06-01T04:40:56.268037Z [debug    ] 1 running task instances for executor KubernetesExecutor [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=base_executor.py:369
2026-06-01T04:40:56.268417Z [debug    ] self.running: {TaskInstanceKey(dag_id='test_k8s_executor_config_legacy', task_id='check_for_failure', run_id='manual__2026-06-01T04:32:45.257995+00:00', try_number=1,
map_index=-1)} [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:278
2026-06-01T04:40:56.270587Z [debug    ] Finding 'running' jobs without a recent heartbeat [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:2945
2026-06-01T04:40:57.273072Z [debug    ] Running SchedulerJobRunner._create_dagruns_for_dags with retries. Try 1 of 3 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=retries.py:100
2026-06-01T04:40:57.277208Z [debug    ] Running DagRun.get_queued_dag_runs_to_set_running with retries. Try 1 of 3 [airflow.models.dagrun.DagRun] loc=retries.py:100
2026-06-01T04:40:57.279487Z [debug    ] Running DagRun.get_running_dag_runs_to_examine with retries. Try 1 of 3 [airflow.models.dagrun.DagRun] loc=retries.py:100
2026-06-01T04:40:57.281452Z [debug    ] Running SchedulerJobRunner._schedule_all_dag_runs with retries. Try 1 of 3 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=retries.py:100
2026-06-01T04:40:57.285548Z [debug    ] number of tis tasks for <DagRun test_k8s_executor_config_legacy @ 2026-06-01 04:32:44+00:00: manual__2026-06-01T04:32:45.257995+00:00, state:running, queued_at: 2026-06-01
04:32:45.264255+00:00. run_type: manual>: 1 task(s) [airflow.models.dagrun.DagRun] loc=dagrun.py:1283
2026-06-01T04:40:57.285661Z [debug    ] number of scheduleable tasks for <DagRun test_k8s_executor_config_legacy @ 2026-06-01 04:32:44+00:00: manual__2026-06-01T04:32:45.257995+00:00, state:running, queued_at:
2026-06-01 04:32:45.264255+00:00. run_type: manual>: 0 task(s) [airflow.models.dagrun.DagRun] loc=dagrun.py:1304
2026-06-01T04:40:57.294507Z [debug    ] 1 running task instances for executor KubernetesExecutor [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=base_executor.py:369
2026-06-01T04:40:57.294922Z [debug    ] self.running: {TaskInstanceKey(dag_id='test_k8s_executor_config_legacy', task_id='check_for_failure', run_id='manual__2026-06-01T04:32:45.257995+00:00', try_number=1,
map_index=-1)} [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:278
2026-06-01T04:40:58.136461Z [debug    ] Running SchedulerJobRunner._create_dagruns_for_dags with retries. Try 1 of 3 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=retries.py:100
2026-06-01T04:40:58.149041Z [debug    ] Running DagRun.get_queued_dag_runs_to_set_running with retries. Try 1 of 3 [airflow.models.dagrun.DagRun] loc=retries.py:100
2026-06-01T04:40:58.155438Z [debug    ] Running DagRun.get_running_dag_runs_to_examine with retries. Try 1 of 3 [airflow.models.dagrun.DagRun] loc=retries.py:100
2026-06-01T04:40:58.157470Z [debug    ] Running SchedulerJobRunner._schedule_all_dag_runs with retries. Try 1 of 3 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=retries.py:100
2026-06-01T04:40:58.162937Z [debug    ] number of tis tasks for <DagRun test_k8s_executor_config_legacy @ 2026-06-01 04:32:44+00:00: manual__2026-06-01T04:32:45.257995+00:00, state:running, queued_at: 2026-06-01
04:32:45.264255+00:00. run_type: manual>: 1 task(s) [airflow.models.dagrun.DagRun] loc=dagrun.py:1283
2026-06-01T04:40:58.163061Z [debug    ] number of scheduleable tasks for <DagRun test_k8s_executor_config_legacy @ 2026-06-01 04:32:44+00:00: manual__2026-06-01T04:32:45.257995+00:00, state:running, queued_at:
2026-06-01 04:32:45.264255+00:00. run_type: manual>: 0 task(s) [airflow.models.dagrun.DagRun] loc=dagrun.py:1304
2026-06-01T04:40:58.173885Z [debug    ] 1 running task instances for executor KubernetesExecutor [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=base_executor.py:369
2026-06-01T04:40:58.174224Z [debug    ] self.running: {TaskInstanceKey(dag_id='test_k8s_executor_config_legacy', task_id='check_for_failure', run_id='manual__2026-06-01T04:32:45.257995+00:00', try_number=1,
map_index=-1)} [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:278

Flow of function call:

KubernetesExecutor._process_workloads() -> KubernetesExecutor.execute_async() -> PodGenerator.from_obj

So what's happening -

• PodGenerator.from_obj callable receives deprecated config object and a TypeError is raised - executor_config: {'pod_override': 'something something', 'KubernetesExecutor': {'config_file': '/some/path/kubeconfig.yaml'}}
• Gets caught in the exception block of the caller: KubernetesExecutor.execute_async() -> self.fail() is triggered -> change_state removes the key from self.running and exits the function call.
• Control is now back to the top level callable: KubernetesExecutor._process_workloads(), wherein, on the very next line, self.running.add(key) is executed unconditionally — regardless of whether execute_async succeeded or failed.
• Now the key is in self.running — the event buffer already has the FAILED event, but when the scheduler processes it, executor.has_task(ti) returns True (because the key is in self.running), so the failure is ignored.

Fix:

• Move the self.running.add(key) to the success path inside KubernetesExecutor.execute_async() so that the key only gets added to the task queue event buffer if configurations were as per expectation.

---

Was generative AI tooling used to co-author this PR?

• Yes (please specify the tool below)
  Generated-by: Myself, Tested-by: Myself, Assisted-by: Claude Code

---

• Read the Pull Request Guidelines for more information. Note: commit author/co-author name and email in commits become permanently public when merged.
• For fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
• When adding dependency, check compliance with the ASF 3rd Party License Policy.
• For significant user-facing changes create newsfragment: {pr_number}.significant.rst, in airflow-core/newsfragments. You can add this file in a follow-up commit after the PR is created so you know the PR number.

---

Important

🛠️ Maintainer triage note for @kunaljubce · by @potiuk · 2026-07-28 17:13 UTC

This draft PR has been inactive since the last triage note, with no response from the author. Closing to keep the queue clean:

• You are welcome to reopen this PR when you resume work, or open a new one addressing the issues previously raised.
• If you pick it back up, please rebase onto the current main branch first.

The ball is in your court — you've been assigned to this PR. Reopen or open a fresh PR once addressed — no rush.

See the Pull Request quality criteria for how to fix each item. There is no rush.

_{Automated triage — may be imperfect; a maintainer takes the next look. We use this two-stage triage process so maintainers' limited time goes to the conversation with you.}

[kunaljubce]

After the change, the failure is instant on UI and the logs:

2026-06-05T06:34:40.251925969Z stdout F 2026-06-05T06:34:40.251769Z [error    ] Invalid executor_config for TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_depr_configs', task_id='check_for_failure', run_id='manual__2026-06-05T06:34:39.956177+00:00', try_number=1, map_index=-1). Executor_config: {'KubernetesExecutor': {'config_file': '/some/path/kubeconfig.yaml'}} [airflow.providers.cncf.kubernetes.executors.kubernetes_executor.KubernetesExecutor] loc=kubernetes_executor.py:221
2026-06-05T06:34:40.252745511Z stdout F 2026-06-05T06:34:40.252642Z [info     ] Received executor event with state failed for task instance TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_depr_configs', task_id='check_for_failure', run_id='manual__2026-06-05T06:34:39.956177+00:00', try_number=1, map_index=-1) [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1262
2026-06-05T06:34:40.260887552Z stdout F 2026-06-05T06:34:40.260697Z [info     ] TaskInstance Finished: dag_id=test_k8s_executor_config_legacy_depr_configs, task_id=check_for_failure, run_id=manual__2026-06-05T06:34:39.956177+00:00, map_index=-1, ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40, run_start_date=None, run_end_date=None, run_duration=None, state=queued, executor=KubernetesExecutor(parallelism=32), executor_state=failed, try_number=1, max_tries=0, pool=default_pool, queue=default, priority_weight=1, operator=PythonOperator, queued_dttm=2026-06-05 06:34:40.248354+00:00, scheduled_dttm=2026-06-05 06:34:40.237597+00:00,queued_by_job_id=4, pid=None [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1354
2026-06-05T06:34:40.261913261Z stdout F 2026-06-05T06:34:40.261694Z [error    ] Executor KubernetesExecutor(parallelism=32) reported that the task instance <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [queued] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40> finished with state failed, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally Extra info: Invalid executor_config passed [airflow.task] loc=taskinstance.py:1800
2026-06-05T06:34:40.264229552Z stdout F 2026-06-05T06:34:40.264106Z [debug    ] Calling 'on_task_instance_failed' with {'previous_state': <TaskInstanceState.RUNNING: 'running'>, 'task_instance': <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [failed] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40>, 'error': "Executor KubernetesExecutor(parallelism=32) reported that the task instance <TaskInstance: test_k8s_executor_config_legacy_depr_configs.check_for_failure manual__2026-06-05T06:34:39.956177+00:00 [queued] ti_id=019e967d-afdc-7505-a071-e6a1c1fe1b40> finished with state failed, but the task instance's state attribute is queued. Learn more: https://airflow.apache.org/docs/apache-airflow/stable/troubleshooting.html#task-state-changed-externally Extra info: Invalid executor_config passed"} [airflow._shared.listeners.listener] loc=listener.py:59
2026-06-05T06:34:40.264410677Z stdout F 2026-06-05T06:34:40.264310Z [info     ] [InformaticaLineageListener] Task: check_for_failure State: failed [airflow.providers.informatica.plugins.listener] loc=listener.py:77
2026-06-05T06:34:40.264469802Z stdout F 2026-06-05T06:34:40.264371Z [info     ] [InformaticaLineageListener] Skipping lineage handling for state: failed [airflow.providers.informatica.plugins.listener] loc=listener.py:80
2026-06-05T06:34:40.264487261Z stdout F 2026-06-05T06:34:40.264415Z [debug    ] Result from 'on_task_instance_failed': [] [airflow._shared.listeners.listener] loc=listener.py:74
2026-06-05T06:34:40.264546802Z stdout F 2026-06-05T06:34:40.264490Z [info     ] Marking task as FAILED. dag_id=test_k8s_executor_config_legacy_depr_configs, task_id=check_for_failure, run_id=manual__2026-06-05T06:34:39.956177+00:00, logical_date=20260605T063438, start_date=, end_date=20260605T063440 [airflow.models.taskinstance] loc=taskinstance.py:1892

[kunaljubce]

Also validated with correct configs and the dag is succeeding as expected:
test_k8s_executor_config_legacy_correct_configs.py

→ cat /var/folders/4y/hpq5jk211d38sg8p7ry2ndd80000gp/T/kind_logs_airflow-python-3.10-v1.30.13/airflow-python-3.10-v1.30.13-worker/containers/airflow-scheduler-794ddcc7b4-pjdzs_airflow_scheduler-58768f7c960f35ef881f2bcca95a2652cfec4ae29cedd64952ec7c60bec857f2.log | grep -i "success\|succeeded"
2026-06-05T07:46:23.535930961Z stdout F 2026-06-05T07:46:23.535867Z [info     ] Marking run <DagRun test_k8s_executor_config_legacy_correct_configs @ 2026-06-05 07:46:04+00:00: manual__2026-06-05T07:46:06.020918+00:00, state:running, queued_at: 2026-06-05 07:46:06.031141+00:00. run_type: manual> successful [airflow.models.dagrun.DagRun] loc=dagrun.py:1205
2026-06-05T07:46:23.536007127Z stdout F 2026-06-05T07:46:23.535951Z [debug    ] Calling 'on_dag_run_success' with {'dag_run': <DagRun test_k8s_executor_config_legacy_correct_configs @ 2026-06-05 07:46:04+00:00: manual__2026-06-05T07:46:06.020918+00:00, state:success, queued_at: 2026-06-05 07:46:06.031141+00:00. run_type: manual>, 'msg': 'success'} [airflow._shared.listeners.listener] loc=listener.py:59
2026-06-05T07:46:23.536059627Z stdout F 2026-06-05T07:46:23.536021Z [debug    ] Result from 'on_dag_run_success': [] [airflow._shared.listeners.listener] loc=listener.py:74
2026-06-05T07:46:23.536108836Z stdout F 2026-06-05T07:46:23.536062Z [info     ] DagRun Finished: dag_id=test_k8s_executor_config_legacy_correct_configs, logical_date=2026-06-05 07:46:04+00:00, run_id=manual__2026-06-05T07:46:06.020918+00:00, run_start_date=2026-06-05 07:46:06.539397+00:00, run_end_date=2026-06-05 07:46:23.535908+00:00, run_duration=16.996511, state=success, run_type=manual, data_interval_start=2026-06-05 07:46:04+00:00, data_interval_end=2026-06-05 07:46:04+00:00, [airflow.models.dagrun.DagRun] loc=dagrun.py:1271
2026-06-05T07:46:25.180544628Z stdout F 2026-06-05T07:46:25.180427Z [info     ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.60857942Z stdout F 2026-06-05T07:46:25.608149Z [debug    ] response body: {"kind":"Pod","apiVersion":"v1","metadata":{"name":"test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa","namespace":"airflow","uid":"111c1643-b231-4b7e-a4e2-
...
2026-06-05T07:46:25.613855003Z stdout F 2026-06-05T07:46:25.613794Z [info     ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.616947003Z stdout F 2026-06-05T07:46:25.616844Z [info     ] Received executor event with state success for task instance TaskInstanceKey(dag_id='test_k8s_executor_config_legacy_correct_configs', task_id='check_for_failure', run_id='manual__2026-06-05T07:46:06.020918+00:00', try_number=1, map_index=-1) [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1262
2026-06-05T07:46:25.619410753Z stdout F 2026-06-05T07:46:25.619321Z [info     ] Event: test-k8s-executor-config-legacy-correct-configs-check-x3vrkcwa Succeeded, annotations: <omitted> [airflow.providers.cncf.kubernetes.executors.kubernetes_executor_utils.KubernetesJobWatcher] loc=kubernetes_executor_utils.py:322
2026-06-05T07:46:25.621444753Z stdout F 2026-06-05T07:46:25.621337Z [info     ] TaskInstance Finished: dag_id=test_k8s_executor_config_legacy_correct_configs, task_id=check_for_failure, run_id=manual__2026-06-05T07:46:06.020918+00:00, map_index=-1, ti_id=019e96bf-165c-73f2-a28c-849a0866dc9f, run_start_date=2026-06-05 07:46:14.614240+00:00, run_end_date=2026-06-05 07:46:23.205383+00:00, run_duration=8.591143, state=success, executor=KubernetesExecutor(parallelism=32), executor_state=success, try_number=1, max_tries=0, pool=default_pool, queue=default, priority_weight=1, operator=PythonOperator, queued_dttm=2026-06-05 07:46:06.613745+00:00, scheduled_dttm=2026-06-05 07:46:06.573257+00:00,queued_by_job_id=3, pid=15 [airflow.jobs.scheduler_job_runner.SchedulerJobRunner] loc=scheduler_job_runner.py:1354

[jscheffl]

I looked at the code and first-hand looks reasonable but (1) I am not sure about side effects putting a task to "running" list even if it just entered the queue - fear there are side-effects - and (2) am not sure about the added raise if this is something still used and need to be kept ignored instead of raising an error.

As not being expert in K8sExecutor and not using it would request some other reviewer with expertise on this: @jedcunningham can you take a look or do you know an expert if the fix is OK like this?

[kunaljubce]

I looked at the code and first-hand looks reasonable but (1) I am not sure about side effects putting a task to "running" list even if it just entered the queue - fear there are side-effects - and (2) am not sure about the added raise if this is something still used and need to be kept ignored instead of raising an error.

As not being expert in K8sExecutor and not using it would request some other reviewer with expertise on this: @jedcunningham can you take a look or do you know an expert if the fix is OK like this?

So regarding (2), from what I see during my testing, a TypeError is still being raised with the same config -

        if isinstance(k8s_object, k8s.V1Pod):
            return k8s_object
        raise TypeError(
            "Cannot convert a non-kubernetes.client.models.V1Pod object into a KubernetesExecutorConfig"
        )

With the raise block I added, the intention is only to raise a more appropriate error msg to highlight that KubernetesExecutor attribute is no longer supported inside the executor_config. This was previously a deprecation warning introduced in #10756 and deprecated entirely thereafter.

I am happy to understand any possible issues with (1) and test it out or try out an alternative approach.

[kunaljubce]

@potiuk With ref. to your automation comments, I have resolved the review comments after pushing in the fixes. Do I need to do anything for the assignment?

[shahar1]

@kunaljubce Could you please resolve conflicts, as well as rename the PR's title so it would be clear what it does?

[kunaljubce]

@kunaljubce Could you please resolve conflicts, as well as rename the PR's title so it would be clear what it does?

I see there are some changes done to kubernetes_executor.py, so I need to revalidate if the issue persists after the change and make the necessary fixes to get the CI to pass. Acknowledging the gap here, I'll be back on this in a couple of days.

[potiuk]

@kunaljubce This PR has been converted to draft because it does not yet meet our Pull Request quality criteria.

Issues found:

• ❌ Pre-commit / static checks: Failing: CI image checks / Static checks. Run prek run --from-ref main --stage pre-commit locally to reproduce and fix.
• ❌ mypy (type checking): Failing: MyPy providers checks. Run the relevant mypy hook locally, e.g. prek run mypy-airflow-core --all-files (providers: breeze run mypy path/to/code).
• ❌ Provider tests: Failing: provider distributions tests / Compat 2.11.1:P3.10:, provider distributions tests / Compat 3.0.6:P3.10:, provider distributions tests / Compat 3.1.8:P3.10:, provider distributions tests / Compat 3.2.2:P3.10:. Run breeze testing providers-tests --test-type "Providers[<provider>]" locally.

Note: Your branch is 297 commits behind main. Some check failures may be caused by changes in the base branch rather than by your PR. Please rebase your branch and push again to get up-to-date CI results.

What to do next:

• Fix each issue listed above.
• Make sure static checks pass locally (prek run --from-ref main --stage pre-commit).
• Mark the PR as "Ready for review" when you're done.

Converting a PR to draft is not a rejection — it is an invitation to bring the PR up to the project's standards so that maintainer review time is spent productively. There is no rush — take your time and work at your own pace. We appreciate your contribution and are happy to wait for updates. If you have questions, feel free to ask on the Airflow Slack.

---

Note: This comment was drafted by an AI-assisted triage tool and may contain mistakes. Once you have addressed the points above, an Apache Airflow maintainer — a real person — will take the next look at your PR. We use this two-stage triage process so that our maintainers' limited time is spent where it matters most: the conversation with you.