fix: ensure Celery tasks are registered at worker startup (main) (apache#63110)

* fix: ensure Celery tasks are registered at worker startup

Fixes: apache#63043

* fix: add test for celery task registration on import

* remove temporary fix scripts

* chore: add TODO comment for execute_command removal when min Airflow >= 3

* fix: correct indentation of TODO comment in test

Author: kalluripradeep

providers/celery/src/airflow/providers/celery/executors/celery_executor.py

@@ -39,6 +39,9 @@
 
 from airflow.exceptions import AirflowProviderDeprecationWarning
 from airflow.executors.base_executor import BaseExecutor
+from airflow.providers.celery.executors import (
+    celery_executor_utils as _celery_executor_utils,  # noqa: F401 # Needed to register Celery tasks at worker startup, see #63043
+)
 from airflow.providers.celery.version_compat import AIRFLOW_V_3_0_PLUS, AIRFLOW_V_3_2_PLUS
 from airflow.providers.common.compat.sdk import AirflowTaskTimeout, Stats
 from airflow.utils.state import TaskInstanceState

providers/celery/tests/unit/celery/executors/test_celery_executor.py

@@ -687,3 +687,26 @@ def test_task_routing_through_team_specific_app(self, mock_send_tasks, monkeypat
             # Critical: task belongs to team A's app, not module-level app
             assert task_from_call.app is team_a_executor.celery_app
             assert task_from_call.name == "execute_command"
+
+
+def test_celery_tasks_registered_on_import():
+    """
+    Ensure execute_workload (and execute_command for Airflow 2.x) are registered
+    with the Celery app when celery_executor is imported.
+
+    Regression test for https://github.com/apache/airflow/issues/63043
+    Celery provider 3.17.0 exposed that celery_executor_utils was never imported
+    at module level, so tasks were never registered at worker startup.
+    """
+    from airflow.providers.celery.executors.celery_executor_utils import app
+
+    registered_tasks = list(app.tasks.keys())
+    assert "execute_workload" in registered_tasks, (
+        "execute_workload must be registered with the Celery app at import time. "
+        "Workers need this to receive tasks without KeyError."
+    )
+    # TODO: remove this block when min supported Airflow version is >= 3.0
+    if not AIRFLOW_V_3_0_PLUS:
+        assert "execute_command" in registered_tasks, (
+            "execute_command must be registered for Airflow 2.x compatibility."
+        )