PrabhodUrs opened a new issue, #75315:
URL: https://github.com/apache/airflow/issues/75315
### Under which category would you file this issue?
Providers
### Apache Airflow version
3.2.2
### What happened and how to reproduce it?
Deadline alerts using `SyncCallback` fail with Airflow 3.2.2 and
`apache-airflow-providers-celery==3.20.0`.
The scheduler queues the callback workload, but a Celery worker rejects it
with an error similar to:
```text
ValueError: CeleryExecutor does not know how to handle
<class 'airflow.executors.workloads.callback.ExecuteCallback'>
```
## Error logs
The Celery worker rejects the callback workload when the deadline is
exceeded:
```text
ValueError: CeleryExecutor does not know how to handle
<class 'airflow.executors.workloads.callback.ExecuteCallback'>
File ".../celery/app/trace.py", line 585, in trace_task
R = retval = fun(*args, **kwargs)
File ".../celery/app/trace.py", line 858, in __protected_call__
return self.run(*args, **kwargs)
File ".../airflow/providers/celery/executors/celery_executor_utils.py",
line 209, in execute_workload
return execute_workload_pre_3_3(inputs)
File ".../airflow/providers/celery/executors/celery_executor_utils.py",
line 271, in _execute_workload_pre_3_3
raise ValueError(...)
```
The version above is pinned by Airflow 3.2.2's `constraints-3.2.2.txt`.
For comparison, the same deadline callback worked correctly with **Apache
Airflow 3.2.0 and `apache-airflow-providers-celery==3.17.1`**, the provider
version specified by the corresponding Airflow 3.2.0 constraints file. The
issue reproduces with the Airflow 3.2.2 constraints combination above.
The issue was also not reproduced after manually downgrading the provider to
`3.18.0` on the scheduler and all workers, without changing the DAG or callback
module.
The callback is not executed. The DAG run/UI does not clearly indicate that
the deadline callback failed, so the alert can be missed without an obvious
indication in the UI.
This happens consistently in my reproduction environment.
## How to reproduce
### 1. Install Airflow
Install Airflow 3.2.2 with Python 3.12, `CeleryExecutor`, and
`apache-airflow-providers-celery==3.20.0` (the provider version in
`constraints-3.2.2.txt`).
### 2. Create a callback module
Create `plugins/poc_deadline_callbacks.py`:
```python
import logging
logger = logging.getLogger(__name__)
def deadline_alert(**kwargs):
logger.error("Deadline missed: %s", kwargs.get("context", {}))
```
### 3. Create a DAG with a deadline
Create `dags/poc_deadline_alerts.py`:
```python
from __future__ import annotations
import time
from datetime import timedelta
import pendulum
from airflow.sdk import (
DAG,
DeadlineAlert,
DeadlineReference,
SyncCallback,
task,
)
with DAG(
dag_id="POC_DEADLINE_ALERTS",
schedule=None,
start_date=pendulum.datetime(2025, 1, 1, tz="UTC"),
catchup=False,
deadline=[
DeadlineAlert(
reference=DeadlineReference.DAGRUN_QUEUED_AT,
interval=timedelta(minutes=2),
callback=SyncCallback(
callback="poc_deadline_callbacks.deadline_alert",
kwargs={"alert_type": "DAG_RUN_RUNNING"},
),
),
],
) as dag:
@task
def slow_task():
time.sleep(60)
slow_task()
```
### 4. Trigger the DAG and inspect worker logs
Trigger the DAG and wait for the two-minute deadline to be exceeded. Inspect
the Celery worker logs for the `ValueError` shown above. The callback's
`Deadline missed` log message is not emitted.
### 5. Test the workaround
Downgrade `apache-airflow-providers-celery` to `3.18.0` on the scheduler and
all workers, without changing the DAG or callback module. In my environment,
the callback executes correctly after this downgrade.
### What you think should happen instead?
The Celery worker should recognize and execute the `ExecuteCallback`
workload and invoke the registered `SyncCallback`. If execution fails, the
failure should be surfaced clearly rather than leaving the deadline alert
unprocessed without a clear indication.
### Operating System
Red Hat Enterprise Linux (AQ9)
### Deployment
Virtualenv installation
### Apache Airflow Provider(s)
celery
### Versions of Apache Airflow Providers
apache-airflow-providers-celery==3.20.0
### Official Helm Chart version
Not Applicable
### Kubernetes Version
_No response_
### Helm Chart configuration
_No response_
### Docker Image customizations
_No response_
### Anything else?
### Suspected cause (not yet confirmed)
My preliminary source-code review suggests that the fallback workload
execution path used with Airflow 3.2.x may no longer handle `ExecuteCallback`.
In provider versions 3.17.1 and 3.18.0, the workload execution path appears
to handle `ExecuteCallback` explicitly. In the provider version where the
regression appears, callback execution seems to have been moved into a
version-gated workload execution path, but the fallback used on Airflow 3.2.x
may not retain the `ExecuteCallback` branch.
This is a hypothesis based on source inspection and should be verified by
maintainers.
A possible fix would be to restore explicit `ExecuteCallback` handling in
the Airflow 3.2.x fallback path, using the callback execution helper available
in that Airflow version. I have not tested a patch, so this is a proposed
direction rather than a confirmed fix.
### Why the workaround is not ideal
Pinning the provider to `3.18.0` restores callback execution in my
environment, but it means foregoing fixes in later provider releases. In
particular, my investigation identified a later fix related to pre-assigned
`external_executor_id` behavior intended to help prevent duplicate execution
after a scheduler crash. Please verify the exact related issue/PR and its
applicability before relying on this detail.
## Deployment details
- Python: 3.12
- Executor: `CeleryExecutor`
- Metadata database: PostgreSQL
- Broker: RabbitMQ
### Are you willing to submit PR?
- [x] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of
Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
