KubernetesExecutor feature may be broken in 3.1.0

Apache Airflow version

3.1.0

If "Other Airflow 2 version" selected, which one?

No response

What happened?

I upgraded from 3.0.3 to 3.1.0, then my jobs started to fail.

In my configuration, I'm using value CeleryExecutor,KubernetesExecutor for AIRFLOW__CORE__EXECUTOR. Previously in 3.0.3 this has been working fine: all TIs will be executed via CeleryExecutor, except those operators where I specified executor='KubernetesExecutor'.

But in 3.1.0, those operators where I specified executor='KubernetesExecutor' always fail, with the error below:

[2025-09-30 15:13:00] INFO - Filling up the DagBag from /tmp/s3_dag_bundle_ri5yvvpv/dags/tenant_test/xd_test.py source=airflow.models.dagbag.DagBag loc=dagbag.py:593
[2025-09-30 15:13:00] ERROR - Failed to bag_dag: /tmp/s3_dag_bundle_ri5yvvpv/dags/tenant_test/xd_test.py source=airflow.models.dagbag.DagBag loc=dagbag.py:518
UnknownExecutorException: Task 'xd_asked_for_another_task' specifies executor 'KubernetesExecutor', which is not available. Make sure it is listed in your [core] executors configuration, or update the task's executor to use one of the configured executors.
File "/usr/local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 513 in _process_modules
File "/usr/local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 156 in _validate_executor_fields
UnknownExecutorException: Unknown executor being loaded: KubernetesExecutor
File "/usr/local/lib/python3.12/site-packages/airflow/models/dagbag.py", line 154 in _validate_executor_fields
File "/usr/local/lib/python3.12/site-packages/airflow/executors/executor_loader.py", line 245 in lookup_executor_name_by_str
[2025-09-30 15:13:00] ERROR - Dag not found during start up dag_id=xd_test bundle=BundleInfo(name='s3-dags', version=None) path=dags/tenant_test/xd_test.py source=task loc=task_runner.py:633
[2025-09-30 15:13:05] WARNING - Process exited abnormally exit_code=1 source=task

Seems it's not recognizing the 2nd executors specified under AIRFLOW__CORE__EXECUTOR.

This is breaking the multi executor feature

UPDATE:

Actually seems it's not the multi executor feature being broken, instead, it's the KubernetesExecutor feature.

In the Pod Template file, if I put value LocalExecutor,KubernetesExecutor for AIRFLOW__CORE__EXECUTOR in the Pod Template, it can work. But it should only require LocalExecutor, according to the doc https://airflow.apache.org/docs/apache-airflow-providers-cncf-kubernetes/stable/kubernetes_executor.html#example-pod-templates

So something seems wrong and broken. This will impact a lot KubernetesExecutor users.

What you think should happen instead?

No response

How to reproduce

The DAG I used

from __future__ import annotations

# [START tutorial]
# [START import_module]
import textwrap
from datetime import datetime, timedelta

# Operators; we need this to operate!
from airflow.providers.standard.operators.bash import BashOperator

# The DAG object; we'll need this to instantiate a DAG
from airflow.sdk import DAG

# [END import_module]


# [START instantiate_dag]
with DAG(
    "xd_test",
    # [START default_args]
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function, # or list of functions
        # 'on_success_callback': some_other_function, # or list of functions
        # 'on_retry_callback': another_function, # or list of functions
        # 'sla_miss_callback': yet_another_function, # or list of functions
        # 'on_skipped_callback': another_function, #or list of functions
        # 'trigger_rule': 'all_success'
    },
    # [END default_args]
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:
    # [END instantiate_dag]

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    # [START basic_task]
    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        depends_on_past=False,
        bash_command="sleep 5",
        retries=3,
    )
    # [END basic_task]

    # [START documentation]
    t1.doc_md = textwrap.dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](https://imgs.xkcd.com/comics/fixing_problems.png)
    **Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    # [END documentation]

    # [START jinja_template]
    templated_command = textwrap.dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id="templated",
        depends_on_past=False,
        bash_command=templated_command,
    )
    # [END jinja_template]

    t4 = BashOperator(
        task_id="xd_asked_for_another_task",
        depends_on_past=False,
        bash_command="echo Hi_dude~",
        executor='KubernetesExecutor'
    )

    t1 >> [t2, t3]
# [END tutorial]

Operating System

NAME="Red Hat Enterprise Linux" VERSION="9.6 (Plow)" ID="rhel" ID_LIKE="fedora" VERSION_ID="9.6" PLATFORM_ID="platform:el9" PRETTY_NAME="Red Hat Enterprise Linux 9.6 (Plow)"

Versions of Apache Airflow Providers

No response

Deployment

Official Apache Airflow Helm Chart

Deployment details

No response

Anything else?

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

@XD-DENG

@XD-DENG I think the doc can be updated, The LocalExecutor in pod_template_file isn't needed: Check: #49433

But I am still not sure what changed between 3.0.3 and 3.1.0 that broke it. Need to find that out

@XD-DENG

@XD-DENG Could you possibly try using airflow.models.dag.DAG in Airflow 3.0.3 -- my hypothesis is it will fail with the same error.

In 3.0.3, there were two DAG classes:

airflow.models.dag.DAG (legacy) - had validate_executor_field() in its validate() method ⚠️
airflow.sdk.DAG (recommended) - did NOT have executor validation ✅

The SDK DAG's validate() method did NOT check executors, so worker pods could parse DAGs without knowing about all executors.

However, if someone used the legacy import in 3.0.3:

from airflow.models import DAG  # <-- Legacy DAG with executor validation

They would have hit the same issue (I think) in 3.0.3! This was a latent bug.

What Changed in 3.1.0

PR #54383 (commit f620bcc40a) removed airflow.models.dag.DAG entirely. Now only airflow.sdk.DAG exists.

Since SDK DAG doesn't have executor validation, a new standalone function _validate_executor_fields() was added/ported over and is now called for all DAGs during parsing:

https://github.com/apache/airflow/blob/3.1.0/airflow-core/src/airflow/models/dagbag.py#L149-L160

def _process_modules(self, filepath, mods, file_last_changed_on_disk):
    from airflow.sdk import DAG  # Only SDK DAG exists now
    # ...
    try:
        dag.validate()
        _validate_executor_fields(dag)  # <-- NEW: validates ALL DAGs now
        self.bag_dag(dag=dag)

@kaxil

Hi @kaxil , when I was running the same DAG code in 3.0.3, I was already using from airflow.sdk import DAG. So your hypothesis doesn't seem right to me.

Please find the full DAG code below (modified on top of an example DAG):

from __future__ import annotations

# [START tutorial]
# [START import_module]
import textwrap
from datetime import datetime, timedelta

# Operators; we need this to operate!
from airflow.providers.standard.operators.bash import BashOperator

# The DAG object; we'll need this to instantiate a DAG
from airflow.sdk import DAG

# [END import_module]


# [START instantiate_dag]
with DAG(
    "xd_test",
    # [START default_args]
    # These args will get passed on to each operator
    # You can override them on a per-task basis during operator initialization
    default_args={
        "depends_on_past": False,
        "retries": 1,
        "retry_delay": timedelta(minutes=5),
        # 'queue': 'bash_queue',
        # 'pool': 'backfill',
        # 'priority_weight': 10,
        # 'end_date': datetime(2016, 1, 1),
        # 'wait_for_downstream': False,
        # 'execution_timeout': timedelta(seconds=300),
        # 'on_failure_callback': some_function, # or list of functions
        # 'on_success_callback': some_other_function, # or list of functions
        # 'on_retry_callback': another_function, # or list of functions
        # 'sla_miss_callback': yet_another_function, # or list of functions
        # 'on_skipped_callback': another_function, #or list of functions
        # 'trigger_rule': 'all_success'
    },
    # [END default_args]
    description="A simple tutorial DAG",
    schedule=timedelta(days=1),
    start_date=datetime(2021, 1, 1),
    catchup=False,
    tags=["example"],
) as dag:
    # [END instantiate_dag]

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    # [START basic_task]
    t1 = BashOperator(
        task_id="print_date",
        bash_command="date",
    )

    t2 = BashOperator(
        task_id="sleep",
        depends_on_past=False,
        bash_command="sleep 5",
        retries=3,
    )
    # [END basic_task]

    # [START documentation]
    t1.doc_md = textwrap.dedent(
        """\
    #### Task Documentation
    You can document your task using the attributes `doc_md` (markdown),
    `doc` (plain text), `doc_rst`, `doc_json`, `doc_yaml` which gets
    rendered in the UI's Task Instance Details page.
    ![img](https://imgs.xkcd.com/comics/fixing_problems.png)
    **Image Credit:** Randall Munroe, [XKCD](https://xkcd.com/license.html)
    """
    )

    dag.doc_md = __doc__  # providing that you have a docstring at the beginning of the DAG; OR
    dag.doc_md = """
    This is a documentation placed anywhere
    """  # otherwise, type it like this
    # [END documentation]

    # [START jinja_template]
    templated_command = textwrap.dedent(
        """
    {% for i in range(5) %}
        echo "{{ ds }}"
        echo "{{ macros.ds_add(ds, 7)}}"
    {% endfor %}
    """
    )

    t3 = BashOperator(
        task_id="templated",
        depends_on_past=False,
        bash_command=templated_command,
    )
    # [END jinja_template]

    t4 = BashOperator(
        task_id="xd_asked_for_another_task",
        depends_on_past=False,
        bash_command="echo Hi_dude~",
        executor='KubernetesExecutor'
    )

    t1 >> [t2, t3]
# [END tutorial]

Hi @kaxil , when I was running the same DAG code in 3.0.3, I was already using from airflow.sdk import DAG. So your hypothesis doesn't seem right to me.

Yeah exactly, I am saying do the opposite.. i.e. use airflow.models.dag.DAG. We were migrating the code slowly from models.DAG to sdk.DAG. sdk.DAG didn't have all the methods in it in 3.0.x but in 3.1 courtesy of #54383 we migrated everything.

@kaxil

I'm sorry @kaxil , but I'm still a bit confused.

So do you think this is a bug needing fix?

I'm using from airflow.sdk import DAG in both 3.0.3 and 3.1.0, and only got issue in 3.1.0.
And according to what you shared above, using from airflow.sdk import DAG is the right thing to do in 3.1.0.

Sorry if I missed or misunderstood anything ;-)

@kaxil

And according to what you shared above, using from airflow.sdk import DAG is the right thing to do in 3.1.0.

Yes. It's correct.

But @kaxil asked you to try to set deprecated airflow.models.dag.DAG in 3.0.3 to confirm his hypothesis. If you set airflow.models.dag.DAG in 3.0.3, it should fail in the same way and that will confirm Kaxil's hypothesis. If you confirm, this will help us to get the fix out in 3.1.1 @XD-DENG.

(Just looking at that conversation briefly while stil on vacations, I hope I can help to clarify it)

And according to what you shared above, using from airflow.sdk import DAG is the right thing to do in 3.1.0.

Yes. It's correct.

But @kaxil asked you to try to set deprecated airflow.models.dag.DAG in 3.0.3 to confirm his hypothesis. If you set airflow.models.dag.DAG in 3.0.3, it should fail in the same way and that will confirm Kaxil's hypothesis. If you confirm, this will help us to get the fix out in 3.1.1 @XD-DENG.

(Just looking at that conversation briefly while stil on vacations, I hope I can help to clarify it)

Precisely, yeah

@XD-DENG

I was having the same error with 3.1.1, and I am not 100% sure but I think that it was due to my podTemplate block in my values.yaml config file. I think I did not touch it since a while, like before migration to Airflow 3.x.x. After commenting my podTemplate block then the error was gone.

In my case, a pod would spin up still, so Airflow was understanding that the KubernetesExecutor should be used, but something was interfering inside the pod (if not wrong) and the podTemplate has the power to do that, like setting environment variable AIRFLOW__CORE__EXECUTOR for example.
I had the following in my podTemplate, so LocalExecutor was the only choice possible in the list, I guess that made sense while using the KubernetesCeleryExecutor??

@XD-DENG : do you use podTemplate in your case?

I run into "Dag not found during start up" error in one AKS cluster but worked in the other testing env.

I have the same error in 3.1.3 but the other way around - I'm using KubernetesExecutor as the default executor in my system so when trying to add a task to Dag that runs with CeleryExecutor I get this issue. Any idea how to handle it?

@kaxil

@kaxil @potiuk any idea how I can overcome this issue? can't use the CeleryExecutor in the same dag as KubernetesExecutor which is my default one

@dor-bernstein

No idea - but I think some fixes regarding executors were done since - so trying 3.1.5 by anyone, and posting exact case and logs from that version would help whoever will look at it @dor-bernstein

@XD-DENG

@XD-DENG -> has your problem been fixed ?

@potiuk

Hi @potiuk , no, I don't think it has been fixed.

A few weeks ago, I tried to use the latest version 3.1.5, and the issue is still there: in order to make tasks run successfully with KubernetesExecutor, I still have to add KubernetesExecutor for AIRFLOW__CORE__EXECUTOR in pod_template_file, while the expectation is to just specify LocalExecutor for AIRFLOW__CORE__EXECUTOR in pod_template_file (this is the behavior in earlier version, as well as the instruction in the docs).

I plan to add a PR for this, but it may take time due to my schedule.

For anyone else who's being impacted by this: the work-around for this issue for now is simple, just add add KubernetesExecutor for AIRFLOW__CORE__EXECUTOR in your pod_template_file.

Add deep analysis files for 3 open Airflow issues

Fix KubernetesExecutor broken by executor validation in worker pods

@XD-DENG

Hi @XD-DENG , is this issue still open and needs a fix? If so, I'd be happy to work on it and submit a PR.

A fix for this is up in #73172 — it's temporarily closed due to Apache Airflow's new open-PR-limit policy, combined with a GitHub-side bug that's currently blocking me from reopening it. Flagging here so it doesn't get duplicated while that gets sorted out.

Drafted-by: Claude Code (Sonnet 5)

Navigation Menu

Repository navigation

KubernetesExecutor feature may be broken in 3.1.0 #56271

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

What Changed in 3.1.0

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

KubernetesExecutor feature may be broken in 3.1.0 #56271

Description

Apache Airflow version

If "Other Airflow 2 version" selected, which one?

What happened?

What you think should happen instead?

How to reproduce

Operating System

Versions of Apache Airflow Providers

Deployment

Deployment details

Anything else?

Are you willing to submit PR?

Code of Conduct

Activity

kaxil commented on Oct 1, 2025

kaxil commented on Oct 1, 2025

What Changed in 3.1.0

XD-DENG commented on Oct 1, 2025

kaxil commented on Oct 1, 2025

XD-DENG commented on Oct 2, 2025

potiuk commented on Oct 3, 2025

kaxil commented on Oct 6, 2025

Ferdinanddb commented on Oct 29, 2025

snowsky commented on Dec 2, 2025

dor-bernstein commented on Dec 18, 2025

dor-bernstein commented on Jan 1, 2026

potiuk commented on Jan 3, 2026

potiuk commented on Jan 3, 2026

XD-DENG commented on Jan 4, 2026

namanjain24-sudo commented on Sep 15, 2026

namanjain24-sudo commented on Sep 25, 2026

Metadata

Metadata

Assignees

Labels

Type

Projects

Milestone

Relationships

Development

Issue actions