(airflow) branch main updated: Scheduler shouldn't crash when 'executor_config' is passed for executors using task SDK (#47548)

[husseinawala]

This is an automated email from the ASF dual-hosted git repository.

husseinawala 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 e35cf2febb3 Scheduler shouldn't crash when 'executor_config' is passed 
for executors using task SDK (#47548)
e35cf2febb3 is described below

commit e35cf2febb3b2814544cd44d7f331c80c3ed7e99
Author: Amogh Desai <amoghrajesh1...@gmail.com>
AuthorDate: Sun Mar 9 21:50:27 2025 +0530

    Scheduler shouldn't crash when 'executor_config' is passed for executors 
using task SDK (#47548)
---
 airflow/executors/workloads.py                                          | 2 +-
 .../src/airflow/providers/celery/executors/celery_executor_utils.py     | 2 +-
 2 files changed, 2 insertions(+), 2 deletions(-)

diff --git a/airflow/executors/workloads.py b/airflow/executors/workloads.py
index 00ca97405a0..f42a0fe062f 100644
--- a/airflow/executors/workloads.py
+++ b/airflow/executors/workloads.py
@@ -61,7 +61,7 @@ class TaskInstance(BaseModel):
     pool_slots: int
     queue: str
     priority_weight: int
-    executor_config: dict | None = None
+    executor_config: dict | None = Field(default=None, exclude=True)
 
     # TODO: Task-SDK: Can we replace TastInstanceKey with just the uuid across 
the codebase?
     @property
diff --git 
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
 
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
index 098adde9c1a..cbe3eec2309 100644
--- 
a/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
+++ 
b/providers/celery/src/airflow/providers/celery/executors/celery_executor_utils.py
@@ -264,7 +264,7 @@ def send_task_to_executor(
     if AIRFLOW_V_3_0_PLUS:
         if TYPE_CHECKING:
             assert isinstance(args, workloads.BaseWorkload)
-        args = (args.model_dump_json(exclude={"ti": {"executor_config"}}),)
+        args = (args.model_dump_json(),)
     try:
         with timeout(seconds=OPERATION_TIMEOUT):
             result = task_to_run.apply_async(args=args, queue=queue)