This is an automated email from the ASF dual-hosted git repository.
eladkal 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 498c1cd470c Fix CeleryExecutor failing to load with Celery 5.7 (#76591)
498c1cd470c is described below
commit 498c1cd470c0e58c4cdf667595b9a76f68b2e484
Author: Lazizbek Ergashev <[email protected]>
AuthorDate: Fri Nov 6 20:41:30 2026 +0500
Fix CeleryExecutor failing to load with Celery 5.7 (#76591)
---
.../providers/celery/executors/celery_executor_utils.py | 12 ++++++++++--
.../celery/tests/integration/celery/test_celery_executor.py | 7 +++----
2 files changed, 13 insertions(+), 6 deletions(-)
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 c1355c2ac55..8e3cfa4adca 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
@@ -38,9 +38,12 @@ from typing import TYPE_CHECKING, Any
from celery import Celery, states as celery_states
from celery.backends.base import BaseKeyValueStoreBackend
-from celery.backends.database import DatabaseBackend, Task as TaskDb, retry,
session_cleanup
+from celery.backends.database import DatabaseBackend, Task as TaskDb,
session_cleanup
from celery.signals import import_modules as celery_import_modules,
worker_ready
from sqlalchemy import select
+from sqlalchemy.exc import DatabaseError, InterfaceError, InvalidRequestError
+from sqlalchemy.orm.exc import StaleDataError
+from tenacity import before_sleep_log, retry, retry_if_exception_type,
stop_after_attempt
from airflow.executors.base_executor import BaseExecutor
from airflow.providers.celery.version_compat import (
@@ -551,7 +554,12 @@ class BulkStateFetcher(LoggingMixin):
return self._prepare_state_and_info_by_task_dict(task_ids,
task_results_by_task_id)
- @retry
+ @retry(
+ retry=retry_if_exception_type((DatabaseError, InterfaceError,
InvalidRequestError, StaleDataError)),
+ stop=stop_after_attempt(3),
+ before_sleep=before_sleep_log(log, logging.WARNING),
+ reraise=True,
+ )
def _query_task_cls_from_db_backend(self, task_ids: set[str], **kwargs):
session = self.celery_app.backend.ResultSession()
task_cls = getattr(self.celery_app.backend, "task_cls", TaskDb)
diff --git a/providers/celery/tests/integration/celery/test_celery_executor.py
b/providers/celery/tests/integration/celery/test_celery_executor.py
index b542626f3fe..71d4d650e69 100644
--- a/providers/celery/tests/integration/celery/test_celery_executor.py
+++ b/providers/celery/tests/integration/celery/test_celery_executor.py
@@ -473,10 +473,9 @@ class TestBulkStateFetcher:
)
assert mock_retry_db_result.call_count == 2
assert result == {"123": ("SUCCESS", None), "456": ("PENDING", None)}
- assert caplog.messages == [
- "Failed operation _query_task_cls_from_db_backend. Retrying 2
more times.",
- "Fetched 2 state(s) for 2 task(s)",
- ]
+ assert len(caplog.messages) == 2
+ assert "_query_task_cls_from_db_backend in 0 seconds as it raised
DatabaseError" in caplog.messages[0]
+ assert caplog.messages[1] == "Fetched 2 state(s) for 2 task(s)"
def test_should_support_base_backend(self, caplog):
from airflow.providers.celery.executors import celery_executor_utils
